KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use setup_driver method in toolium

Best Python code snippet using toolium_python

test_hp3par.py

Source:test_hp3par.py

...324        task_statuses = [self.STATUS_ACTIVE, self.STATUS_ACTIVE]325        def side_effect(*args):326            return task_statuses and task_statuses.pop(0) or self.STATUS_DONE327        conf = {'getTask.side_effect': side_effect}328        mock_client = self.setup_driver(mock_conf=conf)329        task_id = 1234330        interval = .001331        waiter = self.driver.common.TaskWaiter(mock_client, task_id, interval)332        status = waiter.wait_for_task()333        expected = [334            mock.call.getTask(task_id),335            mock.call.getTask(task_id),336            mock.call.getTask(task_id)337        ]338        mock_client.assert_has_calls(expected)339        self.assertEqual(status, self.STATUS_DONE)340    def test_create_volume(self):341        # setup_mock_client drive with default configuration342        # and return the mock HTTP 3PAR client343        mock_client = self.setup_driver()344        self.driver.create_volume(self.volume)345        comment = (346            '{"display_name": "Foo Volume", "type": "OpenStack",'347            ' "name": "volume-d03338a9-9115-48a3-8dfc-35cdfcdc15a7",'348            ' "volume_id": "d03338a9-9115-48a3-8dfc-35cdfcdc15a7"}')349        expected = [350            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),351            mock.call.createVolume(352                self.VOLUME_3PAR_NAME,353                HP3PAR_CPG,354                1907, {355                    'comment': comment,356                    'tpvv': True,357                    'snapCPG': HP3PAR_CPG_SNAP}),358            mock.call.logout()]359        mock_client.assert_has_calls(expected)360    @mock.patch.object(volume_types, 'get_volume_type')361    def test_create_volume_qos(self, _mock_volume_types):362        # setup_mock_client drive with default configuration363        # and return the mock HTTP 3PAR client364        mock_client = self.setup_driver()365        _mock_volume_types.return_value = {366            'name': 'gold',367            'extra_specs': {368                'cpg': HP3PAR_CPG,369                'snap_cpg': HP3PAR_CPG_SNAP,370                'vvs_name': self.VVS_NAME,371                'qos': self.QOS,372                'tpvv': True,373                'volume_type': self.volume_type}}374        self.driver.create_volume(self.volume_qos)375        comment = (376            '{"volume_type_name": "gold", "display_name": "Foo Volume"'377            ', "name": "volume-d03338a9-9115-48a3-8dfc-35cdfcdc15a7'378            '", "volume_type_id": "gold", "volume_id": "d03338a9-91'379            '15-48a3-8dfc-35cdfcdc15a7", "qos": {}, "type": "OpenStack"}')380        expected = [381            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),382            mock.call.createVolume(383                self.VOLUME_3PAR_NAME,384                HP3PAR_CPG,385                1907, {386                    'comment': comment,387                    'tpvv': True,388                    'snapCPG': HP3PAR_CPG_SNAP}),389            mock.call.logout()]390        mock_client.assert_has_calls(expected)391    @mock.patch.object(volume_types, 'get_volume_type')392    def test_retype_not_3par(self, _mock_volume_types):393        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_1394        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)395        self.assertRaises(exception.InvalidHost,396                          self.driver.retype,397                          self.ctxt,398                          self.RETYPE_VOLUME_INFO_0,399                          self.RETYPE_VOLUME_TYPE_1,400                          self.RETYPE_DIFF,401                          self.RETYPE_HOST_NOT3PAR)402        expected = [403            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),404            mock.call.getVolume(self.VOLUME_3PAR_NAME),405            mock.call.logout()]406        mock_client.assert_has_calls(expected)407    @mock.patch.object(volume_types, 'get_volume_type')408    def test_retype_volume_not_found(self, _mock_volume_types):409        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_1410        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)411        mock_client.getVolume.side_effect = hpexceptions.HTTPNotFound412        self.assertRaises(hpexceptions.HTTPNotFound,413                          self.driver.retype,414                          self.ctxt,415                          self.RETYPE_VOLUME_INFO_0,416                          self.RETYPE_VOLUME_TYPE_1,417                          self.RETYPE_DIFF,418                          self.RETYPE_HOST)419        expected = [420            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),421            mock.call.getVolume(self.VOLUME_3PAR_NAME),422            mock.call.logout()]423        mock_client.assert_has_calls(expected)424    @mock.patch.object(volume_types, 'get_volume_type')425    def test_retype_snap_cpg_check(self, _mock_volume_types):426        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_1427        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)428        mock_client.getVolume.return_value = self.RETYPE_VOLUME_INFO_NO_SNAP429        self.assertRaises(exception.InvalidVolume,430                          self.driver.retype,431                          self.ctxt,432                          self.RETYPE_VOLUME_INFO_NO_SNAP,433                          self.RETYPE_VOLUME_TYPE_1,434                          self.RETYPE_DIFF,435                          self.RETYPE_HOST)436        expected = [437            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),438            mock.call.getVolume(self.VOLUME_3PAR_NAME),439            mock.call.getStorageSystemInfo(),440            mock.call.logout()]441        mock_client.assert_has_calls(expected)442    @mock.patch.object(volume_types, 'get_volume_type')443    def test_retype_specs_error_reverts_snap_cpg(self, _mock_volume_types):444        _mock_volume_types.side_effect = [445            self.RETYPE_VOLUME_TYPE_1, self.RETYPE_VOLUME_TYPE_0]446        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)447        mock_client.getVolume.return_value = self.RETYPE_VOLUME_INFO_0448        # Fail the QOS setting to test the revert of the snap CPG rename.449        mock_client.addVolumeToVolumeSet.side_effect = \450            hpexceptions.HTTPForbidden451        self.assertRaises(hpexceptions.HTTPForbidden,452                          self.driver.retype,453                          self.ctxt,454                          {'id': self.VOLUME_ID},455                          self.RETYPE_VOLUME_TYPE_0,456                          self.RETYPE_DIFF,457                          self.RETYPE_HOST)458        old_settings = {459            'snapCPG': self.RETYPE_VOLUME_INFO_0['snapCPG'],460            'comment': self.RETYPE_VOLUME_INFO_0['comment']}461        new_settings = {462            'snapCPG': self.RETYPE_VOLUME_TYPE_1['extra_specs']['snap_cpg'],463            'comment': mock.ANY}464        expected = [465            mock.call.modifyVolume(self.VOLUME_3PAR_NAME, new_settings)466        ]467        mock_client.assert_has_calls(expected)468        expected = [469            mock.call.modifyVolume(self.VOLUME_3PAR_NAME, old_settings),470            mock.call.logout()]471        mock_client.assert_has_calls(expected)472    @mock.patch.object(volume_types, 'get_volume_type')473    def test_retype_revert_comment(self, _mock_volume_types):474        _mock_volume_types.side_effect = [475            self.RETYPE_VOLUME_TYPE_2, self.RETYPE_VOLUME_TYPE_1]476        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)477        mock_client.getVolume.return_value = self.RETYPE_VOLUME_INFO_1478        # Fail the QOS setting to test the revert of the snap CPG rename.479        mock_client.deleteVolumeSet.side_effect = hpexceptions.HTTPForbidden480        self.assertRaises(hpexceptions.HTTPForbidden,481                          self.driver.retype,482                          self.ctxt,483                          {'id': self.VOLUME_ID},484                          self.RETYPE_VOLUME_TYPE_2,485                          self.RETYPE_DIFF,486                          self.RETYPE_HOST)487        original = {488            'snapCPG': self.RETYPE_VOLUME_INFO_1['snapCPG'],489            'comment': self.RETYPE_VOLUME_INFO_1['comment']}490        expected = [491            mock.call.modifyVolume('osv-0DM4qZEVSKON-DXN-NwVpw', original),492            mock.call.logout()]493        mock_client.assert_has_calls(expected)494    @mock.patch.object(volume_types, 'get_volume_type')495    def test_retype_different_array(self, _mock_volume_types):496        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_1497        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)498        mock_client.getStorageSystemInfo.return_value = {499            'serialNumber': 'XXXXXXX'}500        self.assertRaises(exception.InvalidHost,501                          self.driver.retype,502                          self.ctxt,503                          self.RETYPE_VOLUME_INFO_0,504                          self.RETYPE_VOLUME_TYPE_1,505                          self.RETYPE_DIFF,506                          self.RETYPE_HOST)507        expected = [508            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),509            mock.call.getVolume(self.VOLUME_3PAR_NAME),510            mock.call.getStorageSystemInfo(),511            mock.call.logout()]512        mock_client.assert_has_calls(expected)513    @mock.patch.object(volume_types, 'get_volume_type')514    def test_retype_across_cpg_domains(self, _mock_volume_types):515        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_1516        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)517        mock_client.getCPG.side_effect = [518            {'domain': 'domain1'},519            {'domain': 'domain2'},520        ]521        self.assertRaises(exception.Invalid3PARDomain,522                          self.driver.retype,523                          self.ctxt,524                          self.RETYPE_VOLUME_INFO_0,525                          self.RETYPE_VOLUME_TYPE_1,526                          self.RETYPE_DIFF,527                          self.RETYPE_HOST)528        expected = [529            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),530            mock.call.getVolume(self.VOLUME_3PAR_NAME),531            mock.call.getStorageSystemInfo(),532            mock.call.getCPG(self.RETYPE_VOLUME_INFO_0['userCPG']),533            mock.call.getCPG(self.RETYPE_VOLUME_TYPE_1['extra_specs']['cpg']),534            mock.call.logout()535        ]536        mock_client.assert_has_calls(expected)537    @mock.patch.object(volume_types, 'get_volume_type')538    def test_retype_across_snap_cpg_domains(self, _mock_volume_types):539        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_1540        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)541        mock_client.getCPG.side_effect = [542            {'domain': 'cpg_domain'},543            {'domain': 'cpg_domain'},544            {'domain': 'snap_cpg_domain_1'},545            {'domain': 'snap_cpg_domain_2'},546        ]547        self.assertRaises(exception.Invalid3PARDomain,548                          self.driver.retype,549                          self.ctxt,550                          self.RETYPE_VOLUME_INFO_0,551                          self.RETYPE_VOLUME_TYPE_1,552                          self.RETYPE_DIFF,553                          self.RETYPE_HOST)554        expected = [555            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),556            mock.call.getVolume(self.VOLUME_3PAR_NAME),557            mock.call.getStorageSystemInfo(),558            mock.call.getCPG(self.RETYPE_VOLUME_INFO_0['userCPG']),559            mock.call.getCPG(self.RETYPE_VOLUME_TYPE_1['extra_specs']['cpg']),560            mock.call.getCPG(self.RETYPE_VOLUME_INFO_0['snapCPG']),561            mock.call.getCPG(562                self.RETYPE_VOLUME_TYPE_1['extra_specs']['snap_cpg']),563            mock.call.logout()564        ]565        mock_client.assert_has_calls(expected)566    @mock.patch.object(volume_types, 'get_volume_type')567    def test_retype_to_bad_persona(self, _mock_volume_types):568        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_BAD_PERSONA569        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)570        self.assertRaises(exception.InvalidInput,571                          self.driver.retype,572                          self.ctxt,573                          self.RETYPE_VOLUME_INFO_0,574                          self.RETYPE_VOLUME_TYPE_BAD_PERSONA,575                          self.RETYPE_DIFF,576                          self.RETYPE_HOST)577        expected = [578            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),579            mock.call.getVolume(self.VOLUME_3PAR_NAME),580            mock.call.logout()581        ]582        mock_client.assert_has_calls(expected)583    @mock.patch.object(volume_types, 'get_volume_type')584    def test_retype_to_bad_cpg(self, _mock_volume_types):585        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_BAD_CPG586        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)587        mock_client.getCPG.side_effect = hpexceptions.HTTPNotFound588        self.assertRaises(exception.InvalidInput,589                          self.driver.retype,590                          self.ctxt,591                          self.RETYPE_VOLUME_INFO_0,592                          self.RETYPE_VOLUME_TYPE_BAD_CPG,593                          self.RETYPE_DIFF,594                          self.RETYPE_HOST)595        expected = [596            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),597            mock.call.getCPG(598                self.RETYPE_VOLUME_TYPE_BAD_CPG['extra_specs']['cpg']),599            mock.call.logout()600        ]601        mock_client.assert_has_calls(expected)602    @mock.patch.object(volume_types, 'get_volume_type')603    def test_retype_tune(self, _mock_volume_types):604        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_1605        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)606        qos_ref = qos_specs.create(self.ctxt, 'qos-specs-1', self.QOS)607        type_ref = volume_types.create(self.ctxt,608                                       "type1", {"qos:maxIOPS": "100",609                                                 "qos:maxBWS": "50",610                                                 "qos:minIOPS": "10",611                                                 "qos:minBWS": "20",612                                                 "qos:latency": "5",613                                                 "qos:priority": "high"})614        qos_specs.associate_qos_with_type(self.ctxt,615                                          qos_ref['id'],616                                          type_ref['id'])617        type_ref = volume_types.get_volume_type(self.ctxt, type_ref['id'])618        volume = {'id': HP3PARBaseDriver.CLONE_ID}619        self.driver.retype(self.ctxt, volume, type_ref, None, self.RETYPE_HOST)620        expected = [621            mock.call.modifyVolume('osv-0DM4qZEVSKON-AAAAAAAAA',622                                   {'comment': mock.ANY,623                                    'snapCPG': 'OpenStackCPGSnap'}),624            mock.call.deleteVolumeSet('vvs-0DM4qZEVSKON-AAAAAAAAA'),625            mock.call.addVolumeToVolumeSet('myvvs',626                                           'osv-0DM4qZEVSKON-AAAAAAAAA'),627            mock.call.modifyVolume('osv-0DM4qZEVSKON-AAAAAAAAA',628                                   {'action': 6,629                                    'userCPG': 'OpenStackCPG',630                                    'conversionOperation': 1,631                                    'tuneOperation': 1}),632            mock.call.getTask(1),633            mock.call.logout()634        ]635        mock_client.assert_has_calls(expected)636    @mock.patch.object(volume_types, 'get_volume_type')637    def test_retype_qos_spec(self, _mock_volume_types):638        _mock_volume_types.return_value = self.RETYPE_VOLUME_TYPE_1639        mock_client = self.setup_driver(mock_conf=self.RETYPE_CONF)640        cpg = "any_cpg"641        snap_cpg = "any_cpg"642        self.driver.common._retype(self.volume,643                                   HP3PARBaseDriver.VOLUME_3PAR_NAME,644                                   "old_type", "old_type_id",645                                   HP3PARBaseDriver.RETYPE_HOST,646                                   None, cpg, cpg, snap_cpg, snap_cpg,647                                   True, True, None, None,648                                   self.QOS_SPECS, self.RETYPE_QOS_SPECS,649                                   "{}")650        expected = [651            mock.call.createVolumeSet('vvs-0DM4qZEVSKON-DXN-NwVpw', None),652            mock.call.createQoSRules(653                'vvs-0DM4qZEVSKON-DXN-NwVpw',654                {'ioMinGoal': 100, 'ioMaxLimit': 1000,655                 'bwMinGoalKB': 25600, 'bwMaxLimitKB': 51200,656                 'priority': 3,657                 'latencyGoal': 25}658            ),659            mock.call.addVolumeToVolumeSet(660                'vvs-0DM4qZEVSKON-DXN-NwVpw', 'osv-0DM4qZEVSKON-DXN-NwVpw')]661        mock_client.assert_has_calls(expected)662    def test_delete_volume(self):663        # setup_mock_client drive with default configuration664        # and return the mock HTTP 3PAR client665        mock_client = self.setup_driver()666        self.driver.delete_volume(self.volume)667        expected = [668            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),669            mock.call.deleteVolume(self.VOLUME_3PAR_NAME),670            mock.call.logout()]671        mock_client.assert_has_calls(expected)672    def test_create_cloned_volume(self):673        # setup_mock_client drive with default configuration674        # and return the mock HTTP 3PAR client675        mock_client = self.setup_driver()676        mock_client.copyVolume.return_value = {'taskid': 1}677        volume = {'name': HP3PARBaseDriver.VOLUME_NAME,678                  'id': HP3PARBaseDriver.CLONE_ID,679                  'display_name': 'Foo Volume',680                  'size': 2,681                  'host': HP3PARBaseDriver.FAKE_HOST,682                  'source_volid': HP3PARBaseDriver.VOLUME_ID}683        src_vref = {}684        model_update = self.driver.create_cloned_volume(volume, src_vref)685        self.assertIsNotNone(model_update)686        expected = [687            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),688            mock.call.copyVolume(689                self.VOLUME_3PAR_NAME,690                'osv-0DM4qZEVSKON-AAAAAAAAA',691                HP3PAR_CPG,692                {'snapCPG': 'OpenStackCPGSnap', 'tpvv': True,693                 'online': True}),694            mock.call.logout()]695        mock_client.assert_has_calls(expected)696    def test_migrate_volume(self):697        conf = {698            'getStorageSystemInfo.return_value': {699                'serialNumber': '1234'},700            'getTask.return_value': {701                'status': 1},702            'getCPG.return_value': {},703            'copyVolume.return_value': {'taskid': 1},704            'getVolume.return_value': {}705        }706        mock_client = self.setup_driver(mock_conf=conf)707        volume = {'name': HP3PARBaseDriver.VOLUME_NAME,708                  'id': HP3PARBaseDriver.CLONE_ID,709                  'display_name': 'Foo Volume',710                  'size': 2,711                  'status': 'available',712                  'host': HP3PARBaseDriver.FAKE_HOST,713                  'source_volid': HP3PARBaseDriver.VOLUME_ID}714        volume_name_3par = self.driver.common._encode_name(volume['id'])715        loc_info = 'HP3PARDriver:1234:CPG-FC1'716        host = {'host': 'stack@3parfc1',717                'capabilities': {'location_info': loc_info}}718        result = self.driver.migrate_volume(context.get_admin_context(),719                                            volume, host)720        self.assertIsNotNone(result)721        self.assertEqual((True, None), result)722        osv_matcher = 'osv-' + volume_name_3par723        omv_matcher = 'omv-' + volume_name_3par724        expected = [725            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),726            mock.call.getStorageSystemInfo(),727            mock.call.getCPG(HP3PAR_CPG),728            mock.call.getCPG('CPG-FC1'),729            mock.call.copyVolume(osv_matcher, omv_matcher, mock.ANY, mock.ANY),730            mock.call.getTask(mock.ANY),731            mock.call.getVolume(osv_matcher),732            mock.call.deleteVolume(osv_matcher),733            mock.call.modifyVolume(omv_matcher, {'newName': osv_matcher}),734            mock.call.logout()735        ]736        mock_client.assert_has_calls(expected)737    def test_migrate_volume_diff_host(self):738        conf = {739            'getStorageSystemInfo.return_value': {740                'serialNumber': 'different'},741        }742        self.setup_driver(mock_conf=conf)743        volume = {'name': HP3PARBaseDriver.VOLUME_NAME,744                  'id': HP3PARBaseDriver.CLONE_ID,745                  'display_name': 'Foo Volume',746                  'size': 2,747                  'status': 'available',748                  'host': HP3PARBaseDriver.FAKE_HOST,749                  'source_volid': HP3PARBaseDriver.VOLUME_ID}750        loc_info = 'HP3PARDriver:1234:CPG-FC1'751        host = {'host': 'stack@3parfc1',752                'capabilities': {'location_info': loc_info}}753        result = self.driver.migrate_volume(context.get_admin_context(),754                                            volume, host)755        self.assertIsNotNone(result)756        self.assertEqual((False, None), result)757    def test_migrate_volume_diff_domain(self):758        conf = {759            'getStorageSystemInfo.return_value': {760                'serialNumber': '1234'},761            'getTask.return_value': {762                'status': 1},763            'getCPG.side_effect':764            lambda x: {'OpenStackCPG': {'domain': 'OpenStack'}}.get(x, {})765        }766        self.setup_driver(mock_conf=conf)767        volume = {'name': HP3PARBaseDriver.VOLUME_NAME,768                  'id': HP3PARBaseDriver.CLONE_ID,769                  'display_name': 'Foo Volume',770                  'size': 2,771                  'status': 'available',772                  'host': HP3PARBaseDriver.FAKE_HOST,773                  'source_volid': HP3PARBaseDriver.VOLUME_ID}774        loc_info = 'HP3PARDriver:1234:CPG-FC1'775        host = {'host': 'stack@3parfc1',776                'capabilities': {'location_info': loc_info}}777        result = self.driver.migrate_volume(context.get_admin_context(),778                                            volume, host)779        self.assertIsNotNone(result)780        self.assertEqual((False, None), result)781    def test_migrate_volume_attached(self):782        mock_client = self.setup_driver()783        volume = {'name': HP3PARBaseDriver.VOLUME_NAME,784                  'id': HP3PARBaseDriver.CLONE_ID,785                  'display_name': 'Foo Volume',786                  'size': 2,787                  'status': 'in-use',788                  'host': HP3PARBaseDriver.FAKE_HOST,789                  'source_volid': HP3PARBaseDriver.VOLUME_ID}790        volume_name_3par = self.driver.common._encode_name(volume['id'])791        mock_client.getVLUNs.return_value = {792            'members': [{'volumeName': 'osv-' + volume_name_3par}]}793        loc_info = 'HP3PARDriver:1234:CPG-FC1'794        host = {'host': 'stack@3parfc1',795                'capabilities': {'location_info': loc_info}}796        result = self.driver.migrate_volume(context.get_admin_context(),797                                            volume, host)798        self.assertIsNotNone(result)799        self.assertEqual((False, None), result)800    def test_attach_volume(self):801        # setup_mock_client drive with default configuration802        # and return the mock HTTP 3PAR client803        mock_client = self.setup_driver()804        self.driver.attach_volume(context.get_admin_context(),805                                  self.volume,806                                  'abcdef',807                                  'newhost',808                                  '/dev/vdb')809        expected = [810            mock.call.setVolumeMetaData(811                self.VOLUME_3PAR_NAME,812                'HPQ-CS-instance_uuid',813                'abcdef')]814        mock_client.assert_has_calls(expected)815        # test the exception816        mock_client.setVolumeMetaData.side_effect = Exception('Custom ex')817        self.assertRaises(exception.CinderException,818                          self.driver.attach_volume,819                          context.get_admin_context(),820                          self.volume,821                          'abcdef',822                          'newhost',823                          '/dev/vdb')824    def test_detach_volume(self):825        # setup_mock_client drive with default configuration826        # and return the mock HTTP 3PAR client827        mock_client = self.setup_driver()828        self.driver.detach_volume(context.get_admin_context(), self.volume)829        expected = [830            mock.call.removeVolumeMetaData(831                self.VOLUME_3PAR_NAME,832                'HPQ-CS-instance_uuid')]833        mock_client.assert_has_calls(expected)834        # test the exception835        mock_client.removeVolumeMetaData.side_effect = Exception('Custom ex')836        self.assertRaises(exception.CinderException,837                          self.driver.detach_volume,838                          context.get_admin_context(),839                          self.volume)840    def test_create_snapshot(self):841        # setup_mock_client drive with default configuration842        # and return the mock HTTP 3PAR client843        mock_client = self.setup_driver()844        self.driver.create_snapshot(self.snapshot)845        commet = (846            '{"volume_id": "761fc5e5-5191-4ec7-aeba-33e36de44156",'847            ' "display_name": "fakesnap",'848            ' "description": "test description name",'849            ' "volume_name": "volume-d03338a9-9115-48a3-8dfc-35cdfcdc15a7"}')850        expected = [851            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),852            mock.call.createSnapshot(853                'oss-L4I73ONuTci9Fd4ceij-MQ',854                'osv-dh-F5VGRTseuujPjbeRBVg',855                {856                    'comment': commet,857                    'readOnly': True}),858            mock.call.logout()]859        mock_client.assert_has_calls(expected)860    def test_delete_snapshot(self):861        # setup_mock_client drive with default configuration862        # and return the mock HTTP 3PAR client863        mock_client = self.setup_driver()864        self.driver.delete_snapshot(self.snapshot)865        expected = [866            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),867            mock.call.deleteVolume('oss-L4I73ONuTci9Fd4ceij-MQ'),868            mock.call.logout()]869        mock_client.assert_has_calls(expected)870    def test_delete_snapshot_in_use(self):871        # setup_mock_client drive with default configuration872        # and return the mock HTTP 3PAR client873        mock_client = self.setup_driver()874        self.driver.create_snapshot(self.snapshot)875        self.driver.create_volume_from_snapshot(self.volume, self.snapshot)876        ex = hpexceptions.HTTPConflict("In use")877        mock_client.deleteVolume = mock.Mock(side_effect=ex)878        # Deleting the snapshot that a volume is dependent on should fail879        self.assertRaises(exception.SnapshotIsBusy,880                          self.driver.delete_snapshot,881                          self.snapshot)882    def test_delete_snapshot_not_found(self):883        # setup_mock_client drive with default configuration884        # and return the mock HTTP 3PAR client885        mock_client = self.setup_driver()886        self.driver.create_snapshot(self.snapshot)887        try:888            ex = hpexceptions.HTTPNotFound("not found")889            mock_client.deleteVolume = mock.Mock(side_effect=ex)890            self.driver.delete_snapshot(self.snapshot)891        except Exception:892            self.fail("Deleting a snapshot that is missing should act as if "893                      "it worked.")894    def test_create_volume_from_snapshot(self):895        # setup_mock_client drive with default configuration896        # and return the mock HTTP 3PAR client897        mock_client = self.setup_driver()898        self.driver.create_volume_from_snapshot(self.volume, self.snapshot)899        comment = (900            '{"snapshot_id": "2f823bdc-e36e-4dc8-bd15-de1c7a28ff31",'901            ' "display_name": "Foo Volume",'902            ' "volume_id": "d03338a9-9115-48a3-8dfc-35cdfcdc15a7"}')903        expected = [904            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),905            mock.call.createSnapshot(906                self.VOLUME_3PAR_NAME,907                'oss-L4I73ONuTci9Fd4ceij-MQ',908                {909                    'comment': comment,910                    'readOnly': False}),911            mock.call.logout()]912        mock_client.assert_has_calls(expected)913        volume = self.volume.copy()914        volume['size'] = 1915        self.assertRaises(exception.InvalidInput,916                          self.driver.create_volume_from_snapshot,917                          volume, self.snapshot)918    def test_create_volume_from_snapshot_and_extend(self):919        # setup_mock_client drive with default configuration920        # and return the mock HTTP 3PAR client921        conf = {922            'getTask.return_value': {923                'status': 1},924            'copyVolume.return_value': {'taskid': 1},925            'getVolume.return_value': {}926        }927        mock_client = self.setup_driver(mock_conf=conf)928        volume = self.volume.copy()929        volume['size'] = self.volume['size'] + 10930        self.driver.create_volume_from_snapshot(volume, self.snapshot)931        comment = (932            '{"snapshot_id": "2f823bdc-e36e-4dc8-bd15-de1c7a28ff31",'933            ' "display_name": "Foo Volume",'934            ' "volume_id": "d03338a9-9115-48a3-8dfc-35cdfcdc15a7"}')935        volume_name_3par = self.driver.common._encode_name(volume['id'])936        osv_matcher = 'osv-' + volume_name_3par937        omv_matcher = 'omv-' + volume_name_3par938        expected = [939            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),940            mock.call.createSnapshot(941                self.VOLUME_3PAR_NAME,942                'oss-L4I73ONuTci9Fd4ceij-MQ',943                {944                    'comment': comment,945                    'readOnly': False}),946            mock.call.copyVolume(osv_matcher, omv_matcher, mock.ANY, mock.ANY),947            mock.call.getTask(mock.ANY),948            mock.call.getVolume(osv_matcher),949            mock.call.deleteVolume(osv_matcher),950            mock.call.modifyVolume(omv_matcher, {'newName': osv_matcher}),951            mock.call.growVolume(osv_matcher, 10 * 1024),952            mock.call.logout()]953        mock_client.assert_has_calls(expected)954    def test_create_volume_from_snapshot_and_extend_copy_fail(self):955        # setup_mock_client drive with default configuration956        # and return the mock HTTP 3PAR client957        conf = {958            'getTask.return_value': {959                'status': 4,960                'failure message': 'out of disk space'},961            'copyVolume.return_value': {'taskid': 1},962            'getVolume.return_value': {}963        }964        self.setup_driver(mock_conf=conf)965        volume = self.volume.copy()966        volume['size'] = self.volume['size'] + 10967        self.assertRaises(exception.CinderException,968                          self.driver.create_volume_from_snapshot,969                          volume, self.snapshot)970    @mock.patch.object(volume_types, 'get_volume_type')971    def test_create_volume_from_snapshot_qos(self, _mock_volume_types):972        # setup_mock_client drive with default configuration973        # and return the mock HTTP 3PAR client974        mock_client = self.setup_driver()975        _mock_volume_types.return_value = {976            'name': 'gold',977            'extra_specs': {978                'cpg': HP3PAR_CPG,979                'snap_cpg': HP3PAR_CPG_SNAP,980                'vvs_name': self.VVS_NAME,981                'qos': self.QOS,982                'tpvv': True,983                'volume_type': self.volume_type}}984        self.driver.create_volume_from_snapshot(self.volume_qos, self.snapshot)985        comment = (986            '{"snapshot_id": "2f823bdc-e36e-4dc8-bd15-de1c7a28ff31",'987            ' "display_name": "Foo Volume",'988            ' "volume_id": "d03338a9-9115-48a3-8dfc-35cdfcdc15a7"}')989        expected = [990            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),991            mock.call.createSnapshot(992                self.VOLUME_3PAR_NAME,993                'oss-L4I73ONuTci9Fd4ceij-MQ', {994                    'comment': comment,995                    'readOnly': False}),996            mock.call.logout()]997        mock_client.assert_has_calls(expected)998        volume = self.volume.copy()999        volume['size'] = 11000        self.assertRaises(exception.InvalidInput,1001                          self.driver.create_volume_from_snapshot,1002                          volume, self.snapshot)1003    def test_terminate_connection(self):1004        # setup_mock_client drive with default configuration1005        # and return the mock HTTP 3PAR client1006        mock_client = self.setup_driver()1007        mock_client.getHostVLUNs.return_value = [1008            {'active': True,1009             'volumeName': self.VOLUME_3PAR_NAME,1010             'lun': None, 'type': 0}]1011        self.driver.terminate_connection(1012            self.volume,1013            self.connector,1014            force=True)1015        expected = [1016            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1017            mock.call.getHostVLUNs(self.FAKE_HOST),1018            mock.call.deleteVLUN(1019                self.VOLUME_3PAR_NAME,1020                None,1021                self.FAKE_HOST),1022            mock.call.deleteHost(self.FAKE_HOST),1023            mock.call.logout()]1024        mock_client.assert_has_calls(expected)1025    def test_update_volume_key_value_pair(self):1026        # setup_mock_client drive with default configuration1027        # and return the mock HTTP 3PAR client1028        mock_client = self.setup_driver()1029        key = 'a'1030        value = 'b'1031        self.driver.common.update_volume_key_value_pair(1032            self.volume,1033            key,1034            value)1035        expected = [1036            mock.call.setVolumeMetaData(self.VOLUME_3PAR_NAME, key, value)]1037        mock_client.assert_has_calls(expected)1038        # check exception1039        mock_client.setVolumeMetaData.side_effect = Exception('fake')1040        self.assertRaises(exception.VolumeBackendAPIException,1041                          self.driver.common.update_volume_key_value_pair,1042                          self.volume,1043                          None,1044                          'b')1045    def test_clear_volume_key_value_pair(self):1046        # setup_mock_client drive with default configuration1047        # and return the mock HTTP 3PAR client1048        mock_client = self.setup_driver()1049        key = 'a'1050        self.driver.common.clear_volume_key_value_pair(self.volume, key)1051        expected = [1052            mock.call.removeVolumeMetaData(self.VOLUME_3PAR_NAME, key)]1053        mock_client.assert_has_calls(expected)1054        # check the exception1055        mock_client.removeVolumeMetaData.side_effect = Exception('fake')1056        self.assertRaises(exception.VolumeBackendAPIException,1057                          self.driver.common.clear_volume_key_value_pair,1058                          self.volume,1059                          None)1060    def test_extend_volume(self):1061        # setup_mock_client drive with default configuration1062        # and return the mock HTTP 3PAR client1063        mock_client = self.setup_driver()1064        grow_size = 31065        old_size = self.volume['size']1066        new_size = old_size + grow_size1067        self.driver.extend_volume(self.volume, str(new_size))1068        growth_size_mib = grow_size * units.Ki1069        expected = [1070            mock.call.growVolume(self.VOLUME_3PAR_NAME, growth_size_mib)]1071        mock_client.assert_has_calls(expected)1072    def test_extend_volume_non_base(self):1073        extend_ex = hpexceptions.HTTPForbidden(error={'code': 150})1074        conf = {1075            'getTask.return_value': {1076                'status': 1},1077            'getCPG.return_value': {},1078            'copyVolume.return_value': {'taskid': 1},1079            'getVolume.return_value': {},1080            # Throw an exception first time only1081            'growVolume.side_effect': [extend_ex,1082                                       None],1083        }1084        mock_client = self.setup_driver(mock_conf=conf)1085        grow_size = 31086        old_size = self.volume['size']1087        new_size = old_size + grow_size1088        self.driver.extend_volume(self.volume, str(new_size))1089        self.assertEqual(2, mock_client.growVolume.call_count)1090    def test_extend_volume_non_base_failure(self):1091        extend_ex = hpexceptions.HTTPForbidden(error={'code': 150})1092        conf = {1093            'getTask.return_value': {1094                'status': 1},1095            'getCPG.return_value': {},1096            'copyVolume.return_value': {'taskid': 1},1097            'getVolume.return_value': {},1098            # Always fail1099            'growVolume.side_effect': extend_ex1100        }1101        self.setup_driver(mock_conf=conf)1102        grow_size = 31103        old_size = self.volume['size']1104        new_size = old_size + grow_size1105        self.assertRaises(hpexceptions.HTTPForbidden,1106                          self.driver.extend_volume,1107                          self.volume,1108                          str(new_size))1109    def test_get_ports(self):1110        # setup_mock_client drive with default configuration1111        # and return the mock HTTP 3PAR client1112        mock_client = self.setup_driver()1113        mock_client.getPorts.return_value = {1114            'members': [1115                {'portPos': {'node': 0, 'slot': 8, 'cardPort': 2},1116                 'protocol': 2,1117                 'IPAddr': '10.10.120.252',1118                 'linkState': 4,1119                 'device': [],1120                 'iSCSIName': 'iqn.2000-05.com.3pardata:21810002ac00383d',1121                 'mode': 2,1122                 'HWAddr': '2C27D75375D2',1123                 'type': 8},1124                {'portPos': {'node': 1, 'slot': 8, 'cardPort': 1},1125                 'protocol': 2,1126                 'IPAddr': '10.10.220.253',1127                 'linkState': 4,1128                 'device': [],1129                 'iSCSIName': 'iqn.2000-05.com.3pardata:21810002ac00383d',1130                 'mode': 2,1131                 'HWAddr': '2C27D75375D6',1132                 'type': 8},1133                {'portWWN': '20210002AC00383D',1134                 'protocol': 1,1135                 'linkState': 4,1136                 'mode': 2,1137                 'device': ['cage2'],1138                 'nodeWWN': '20210002AC00383D',1139                 'type': 2,1140                 'portPos': {'node': 0, 'slot': 6, 'cardPort': 3}}]}1141        ports = self.driver.common.get_ports()['members']1142        self.assertEqual(len(ports), 3)1143    def test_get_by_qos_spec_with_scoping(self):1144        self.setup_driver()1145        qos_ref = qos_specs.create(self.ctxt, 'qos-specs-1', self.QOS)1146        type_ref = volume_types.create(self.ctxt,1147                                       "type1", {"qos:maxIOPS": "100",1148                                                 "qos:maxBWS": "50",1149                                                 "qos:minIOPS": "10",1150                                                 "qos:minBWS": "20",1151                                                 "qos:latency": "5",1152                                                 "qos:priority": "high"})1153        qos_specs.associate_qos_with_type(self.ctxt,1154                                          qos_ref['id'],1155                                          type_ref['id'])1156        type_ref = volume_types.get_volume_type(self.ctxt, type_ref['id'])1157        qos = self.driver.common._get_qos_by_volume_type(type_ref)1158        self.assertEqual(qos, {'maxIOPS': '1000', 'maxBWS': '50',1159                               'minIOPS': '100', 'minBWS': '25',1160                               'latency': '25', 'priority': 'low'})1161    def test_get_by_qos_spec(self):1162        self.setup_driver()1163        qos_ref = qos_specs.create(self.ctxt, 'qos-specs-1', self.QOS_SPECS)1164        type_ref = volume_types.create(self.ctxt,1165                                       "type1", {"qos:maxIOPS": "100",1166                                                 "qos:maxBWS": "50",1167                                                 "qos:minIOPS": "10",1168                                                 "qos:minBWS": "20",1169                                                 "qos:latency": "5",1170                                                 "qos:priority": "high"})1171        qos_specs.associate_qos_with_type(self.ctxt,1172                                          qos_ref['id'],1173                                          type_ref['id'])1174        type_ref = volume_types.get_volume_type(self.ctxt, type_ref['id'])1175        qos = self.driver.common._get_qos_by_volume_type(type_ref)1176        self.assertEqual(qos, {'maxIOPS': '1000', 'maxBWS': '50',1177                               'minIOPS': '100', 'minBWS': '25',1178                               'latency': '25', 'priority': 'low'})1179    def test_get_by_qos_by_type_only(self):1180        self.setup_driver()1181        type_ref = volume_types.create(self.ctxt,1182                                       "type1", {"qos:maxIOPS": "100",1183                                                 "qos:maxBWS": "50",1184                                                 "qos:minIOPS": "10",1185                                                 "qos:minBWS": "20",1186                                                 "qos:latency": "5",1187                                                 "qos:priority": "high"})1188        type_ref = volume_types.get_volume_type(self.ctxt, type_ref['id'])1189        qos = self.driver.common._get_qos_by_volume_type(type_ref)1190        self.assertEqual(qos, {'maxIOPS': '100', 'maxBWS': '50',1191                               'minIOPS': '10', 'minBWS': '20',1192                               'latency': '5', 'priority': 'high'})1193    def test_create_vlun(self):1194        host = 'fake-host'1195        lun_id = 111196        nsp = '1:2:3'1197        mock_client = self.setup_driver()1198        location = ("%(name)s,%(lunid)s,%(host)s,%(nsp)s" %1199                    {'name': self.VOLUME_NAME,1200                     'lunid': lun_id,1201                     'host': host,1202                     'nsp': nsp})1203        mock_client.createVLUN.return_value = location1204        expected_info = {'volume_name': self.VOLUME_NAME,1205                         'lun_id': lun_id,1206                         'host_name': host,1207                         'nsp': nsp}1208        vlun_info = self.driver.common._create_3par_vlun(self.VOLUME_NAME,1209                                                         host, nsp)1210        self.assertEqual(expected_info, vlun_info)1211        location = ("%(name)s,%(lunid)s,%(host)s" %1212                    {'name': self.VOLUME_NAME,1213                     'lunid': lun_id,1214                     'host': host})1215        mock_client.createVLUN.return_value = location1216        expected_info = {'volume_name': self.VOLUME_NAME,1217                         'lun_id': lun_id,1218                         'host_name': host}1219        vlun_info = self.driver.common._create_3par_vlun(self.VOLUME_NAME,1220                                                         host, None)1221        self.assertEqual(expected_info, vlun_info)1222    @mock.patch.object(volume_types, 'get_volume_type')1223    def test_manage_existing(self, _mock_volume_types):1224        mock_client = self.setup_driver()1225        _mock_volume_types.return_value = {1226            'name': 'gold',1227            'extra_specs': {1228                'cpg': HP3PAR_CPG,1229                'snap_cpg': HP3PAR_CPG_SNAP,1230                'vvs_name': self.VVS_NAME,1231                'qos': self.QOS,1232                'tpvv': True,1233                'volume_type': self.volume_type}}1234        comment = (1235            '{"display_name": "Foo Volume"}')1236        new_comment = (1237            '{"volume_type_name": "gold",'1238            ' "display_name": "Foo Volume",'1239            ' "name": "volume-007dbfce-7579-40bc-8f90-a20b3902283e",'1240            ' "volume_type_id": "acfa9fa4-54a0-4340-a3d8-bfcf19aea65e",'1241            ' "volume_id": "007dbfce-7579-40bc-8f90-a20b3902283e",'1242            ' "qos": {},'1243            ' "type": "OpenStack"}')1244        volume = {'display_name': None,1245                  'volume_type': 'gold',1246                  'volume_type_id': 'acfa9fa4-54a0-4340-a3d8-bfcf19aea65e',1247                  'id': '007dbfce-7579-40bc-8f90-a20b3902283e'}1248        mock_client.getVolume.return_value = {'comment': comment}1249        unm_matcher = self.driver.common._get_3par_unm_name(self.volume['id'])1250        osv_matcher = self.driver.common._get_3par_vol_name(volume['id'])1251        existing_ref = {'source-name': unm_matcher}1252        obj = self.driver.manage_existing(volume, existing_ref)1253        expected_obj = {'display_name': 'Foo Volume'}1254        expected = [1255            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1256            mock.call.getVolume(existing_ref['source-name']),1257            mock.call.modifyVolume(existing_ref['source-name'],1258                                   {'newName': osv_matcher,1259                                    'comment': new_comment}),1260            mock.call.logout()1261        ]1262        mock_client.assert_has_calls(expected)1263        self.assertEqual(expected_obj, obj)1264        volume['display_name'] = 'Test Volume'1265        obj = self.driver.manage_existing(volume, existing_ref)1266        expected_obj = {'display_name': 'Test Volume'}1267        expected = [1268            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1269            mock.call.getVolume(existing_ref['source-name']),1270            mock.call.modifyVolume(existing_ref['source-name'],1271                                   {'newName': osv_matcher,1272                                    'comment': new_comment}),1273            mock.call.logout()1274        ]1275        mock_client.assert_has_calls(expected)1276        self.assertEqual(expected_obj, obj)1277    def test_manage_existing_no_volume_type(self):1278        mock_client = self.setup_driver()1279        comment = (1280            '{"display_name": "Foo Volume"}')1281        new_comment = (1282            '{"type": "OpenStack",'1283            ' "display_name": "Foo Volume",'1284            ' "name": "volume-007dbfce-7579-40bc-8f90-a20b3902283e",'1285            ' "volume_id": "007dbfce-7579-40bc-8f90-a20b3902283e"}')1286        volume = {'display_name': None,1287                  'volume_type': None,1288                  'id': '007dbfce-7579-40bc-8f90-a20b3902283e'}1289        mock_client.getVolume.return_value = {'comment': comment}1290        unm_matcher = self.driver.common._get_3par_unm_name(self.volume['id'])1291        osv_matcher = self.driver.common._get_3par_vol_name(volume['id'])1292        existing_ref = {'source-name': unm_matcher}1293        obj = self.driver.manage_existing(volume, existing_ref)1294        expected_obj = {'display_name': 'Foo Volume'}1295        expected = [1296            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1297            mock.call.getVolume(existing_ref['source-name']),1298            mock.call.modifyVolume(existing_ref['source-name'],1299                                   {'newName': osv_matcher,1300                                    'comment': new_comment}),1301            mock.call.logout()1302        ]1303        mock_client.assert_has_calls(expected)1304        self.assertEqual(expected_obj, obj)1305        volume['display_name'] = 'Test Volume'1306        obj = self.driver.manage_existing(volume, existing_ref)1307        expected_obj = {'display_name': 'Test Volume'}1308        expected = [1309            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1310            mock.call.getVolume(existing_ref['source-name']),1311            mock.call.modifyVolume(existing_ref['source-name'],1312                                   {'newName': osv_matcher,1313                                    'comment': new_comment}),1314            mock.call.logout()1315        ]1316        mock_client.assert_has_calls(expected)1317        self.assertEqual(expected_obj, obj)1318        mock_client.getVolume.return_value = {}1319        volume['display_name'] = None1320        obj = self.driver.manage_existing(volume, existing_ref)1321        expected_obj = {'display_name': None}1322        expected = [1323            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1324            mock.call.getVolume(existing_ref['source-name']),1325            mock.call.modifyVolume(existing_ref['source-name'],1326                                   {'newName': osv_matcher,1327                                    'comment': new_comment}),1328            mock.call.logout()1329        ]1330        mock_client.assert_has_calls(expected)1331        self.assertEqual(expected_obj, obj)1332    def test_manage_existing_invalid_input(self):1333        mock_client = self.setup_driver()1334        volume = {'display_name': None,1335                  'volume_type': None,1336                  'id': '007dbfce-7579-40bc-8f90-a20b3902283e'}1337        mock_client.getVolume.side_effect = hpexceptions.HTTPNotFound('fake')1338        unm_matcher = self.driver.common._get_3par_unm_name(self.volume['id'])1339        existing_ref = {'source-name': unm_matcher}1340        self.assertRaises(exception.InvalidInput,1341                          self.driver.manage_existing,1342                          volume=volume,1343                          existing_ref=existing_ref)1344        expected = [1345            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1346            mock.call.getVolume(existing_ref['source-name']),1347            mock.call.logout()1348        ]1349        mock_client.assert_has_calls(expected)1350    def test_manage_existing_volume_type_exception(self):1351        mock_client = self.setup_driver()1352        comment = (1353            '{"display_name": "Foo Volume"}')1354        volume = {'display_name': None,1355                  'volume_type': 'gold',1356                  'volume_type_id': 'bcfa9fa4-54a0-4340-a3d8-bfcf19aea65e',1357                  'id': '007dbfce-7579-40bc-8f90-a20b3902283e'}1358        mock_client.getVolume.return_value = {'comment': comment}1359        unm_matcher = self.driver.common._get_3par_unm_name(self.volume['id'])1360        existing_ref = {'source-name': unm_matcher}1361        self.assertRaises(exception.ManageExistingVolumeTypeMismatch,1362                          self.driver.manage_existing,1363                          volume=volume,1364                          existing_ref=existing_ref)1365        expected = [1366            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1367            mock.call.getVolume(existing_ref['source-name']),1368            mock.call.logout()1369        ]1370        mock_client.assert_has_calls(expected)1371    def test_manage_existing_get_size(self):1372        mock_client = self.setup_driver()1373        mock_client.getVolume.return_value = {'sizeMiB': 2048}1374        unm_matcher = self.driver.common._get_3par_unm_name(self.volume['id'])1375        volume = {}1376        existing_ref = {'source-name': unm_matcher}1377        size = self.driver.manage_existing_get_size(volume, existing_ref)1378        expected_size = 21379        expected = [1380            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1381            mock.call.getVolume(existing_ref['source-name']),1382            mock.call.logout()1383        ]1384        mock_client.assert_has_calls(expected, True)1385        self.assertEqual(expected_size, size)1386    def test_manage_existing_get_size_invalid_reference(self):1387        mock_client = self.setup_driver()1388        volume = {}1389        existing_ref = {'source-name': self.VOLUME_3PAR_NAME}1390        self.assertRaises(exception.ManageExistingInvalidReference,1391                          self.driver.manage_existing_get_size,1392                          volume=volume,1393                          existing_ref=existing_ref)1394        expected = [1395            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1396            mock.call.logout()1397        ]1398        mock_client.assert_has_calls(expected)1399        existing_ref = {}1400        self.assertRaises(exception.ManageExistingInvalidReference,1401                          self.driver.manage_existing_get_size,1402                          volume=volume,1403                          existing_ref=existing_ref)1404        expected = [1405            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1406            mock.call.logout()1407        ]1408        mock_client.assert_has_calls(expected)1409    def test_manage_existing_get_size_invalid_input(self):1410        mock_client = self.setup_driver()1411        mock_client.getVolume.side_effect = hpexceptions.HTTPNotFound('fake')1412        unm_matcher = self.driver.common._get_3par_unm_name(self.volume['id'])1413        volume = {}1414        existing_ref = {'source-name': unm_matcher}1415        self.assertRaises(exception.InvalidInput,1416                          self.driver.manage_existing_get_size,1417                          volume=volume,1418                          existing_ref=existing_ref)1419        expected = [1420            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1421            mock.call.getVolume(existing_ref['source-name']),1422            mock.call.logout()1423        ]1424        mock_client.assert_has_calls(expected)1425    def test_unmanage(self):1426        mock_client = self.setup_driver()1427        self.driver.unmanage(self.volume)1428        osv_matcher = self.driver.common._get_3par_vol_name(self.volume['id'])1429        unm_matcher = self.driver.common._get_3par_unm_name(self.volume['id'])1430        expected = [1431            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1432            mock.call.modifyVolume(osv_matcher, {'newName': unm_matcher}),1433            mock.call.logout()]1434        mock_client.assert_has_calls(expected)1435class TestHP3PARFCDriver(HP3PARBaseDriver, test.TestCase):1436    properties = {1437        'driver_volume_type': 'fibre_channel',1438        'data': {1439            'target_lun': 90,1440            'target_wwn': ['0987654321234', '123456789000987'],1441            'target_discovered': True,1442            'initiator_target_map': {'123456789012345':1443                                     ['0987654321234', '123456789000987'],1444                                     '123456789054321':1445                                     ['0987654321234', '123456789000987'],1446                                     }}}1447    def setup_driver(self, config=None, mock_conf=None):1448        self.ctxt = context.get_admin_context()1449        mock_client = self.setup_mock_client(1450            conf=config,1451            m_conf=mock_conf,1452            driver=hpfcdriver.HP3PARFCDriver)1453        expected = [1454            mock.call.setSSHOptions(1455                HP3PAR_SAN_IP,1456                HP3PAR_USER_NAME,1457                HP3PAR_USER_PASS,1458                privatekey=HP3PAR_SAN_SSH_PRIVATE,1459                port=HP3PAR_SAN_SSH_PORT,1460                conn_timeout=HP3PAR_SAN_SSH_CON_TIMEOUT),1461            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1462            mock.call.getCPG(HP3PAR_CPG),1463            mock.call.logout()]1464        mock_client.assert_has_calls(expected)1465        mock_client.reset_mock()1466        return mock_client1467    def test_initialize_connection(self):1468        # setup_mock_client drive with default configuration1469        # and return the mock HTTP 3PAR client1470        mock_client = self.setup_driver()1471        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1472        mock_client.getCPG.return_value = {}1473        mock_client.getHost.side_effect = [1474            hpexceptions.HTTPNotFound('fake'),1475            {'name': self.FAKE_HOST,1476                'FCPaths': [{'driverVersion': None,1477                             'firmwareVersion': None,1478                             'hostSpeed': 0,1479                             'model': None,1480                             'portPos': {'cardPort': 1, 'node': 1,1481                                         'slot': 2},1482                             'vendor': None,1483                             'wwn': self.wwn[0]},1484                            {'driverVersion': None,1485                             'firmwareVersion': None,1486                             'hostSpeed': 0,1487                             'model': None,1488                             'portPos': {'cardPort': 1, 'node': 0,1489                                         'slot': 2},1490                             'vendor': None,1491                             'wwn': self.wwn[1]}]}]1492        mock_client.findHost.return_value = self.FAKE_HOST1493        mock_client.getHostVLUNs.return_value = [1494            {'active': True,1495             'volumeName': self.VOLUME_3PAR_NAME,1496             'lun': 90, 'type': 0}]1497        location = ("%(volume_name)s,%(lun_id)s,%(host)s,%(nsp)s" %1498                    {'volume_name': self.VOLUME_3PAR_NAME,1499                     'lun_id': 90,1500                     'host': self.FAKE_HOST,1501                     'nsp': 'something'})1502        mock_client.createVLUN.return_value = location1503        result = self.driver.initialize_connection(self.volume, self.connector)1504        expected = [1505            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1506            mock.call.getVolume(self.VOLUME_3PAR_NAME),1507            mock.call.getCPG(HP3PAR_CPG),1508            mock.call.getHost(self.FAKE_HOST),1509            mock.ANY,1510            mock.call.getHost(self.FAKE_HOST),1511            mock.call.createVLUN(1512                self.VOLUME_3PAR_NAME,1513                auto=True,1514                hostname=self.FAKE_HOST),1515            mock.call.getHostVLUNs(self.FAKE_HOST),1516            mock.call.getPorts(),1517            mock.call.logout()]1518        mock_client.assert_has_calls(expected)1519        self.assertDictMatch(result, self.properties)1520    def test_terminate_connection(self):1521        # setup_mock_client drive with default configuration1522        # and return the mock HTTP 3PAR client1523        mock_client = self.setup_driver()1524        effects = [1525            [{'active': True, 'volumeName': self.VOLUME_3PAR_NAME,1526              'lun': None, 'type': 0}],1527            hpexceptions.HTTPNotFound]1528        mock_client.getHostVLUNs.side_effect = effects1529        expected = [1530            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1531            mock.call.getHostVLUNs(self.FAKE_HOST),1532            mock.call.deleteVLUN(1533                self.VOLUME_3PAR_NAME,1534                None,1535                self.FAKE_HOST),1536            mock.call.deleteHost(self.FAKE_HOST),1537            mock.call.getHostVLUNs(self.FAKE_HOST),1538            mock.call.getPorts(),1539            mock.call.logout()]1540        conn_info = self.driver.terminate_connection(self.volume,1541                                                     self.connector)1542        mock_client.assert_has_calls(expected)1543        self.assertIn('data', conn_info)1544        self.assertIn('initiator_target_map', conn_info['data'])1545        mock_client.reset_mock()1546        mock_client.getHostVLUNs.side_effect = effects1547        # mock some deleteHost exceptions that are handled1548        delete_with_vlun = hpexceptions.HTTPConflict(1549            error={'message': "has exported VLUN"})1550        delete_with_hostset = hpexceptions.HTTPConflict(1551            error={'message': "host is a member of a set"})1552        mock_client.deleteHost = mock.Mock(1553            side_effect=[delete_with_vlun, delete_with_hostset])1554        conn_info = self.driver.terminate_connection(self.volume,1555                                                     self.connector)1556        mock_client.assert_has_calls(expected)1557        mock_client.reset_mock()1558        mock_client.getHostVLUNs.side_effect = effects1559        conn_info = self.driver.terminate_connection(self.volume,1560                                                     self.connector)1561        mock_client.assert_has_calls(expected)1562    def test_terminate_connection_more_vols(self):1563        mock_client = self.setup_driver()1564        # mock more than one vlun on the host (don't even try to remove host)1565        mock_client.getHostVLUNs.return_value = \1566            [1567                {'active': True,1568                 'volumeName': self.VOLUME_3PAR_NAME,1569                 'lun': None, 'type': 0},1570                {'active': True,1571                 'volumeName': 'there-is-another-volume',1572                 'lun': None, 'type': 0},1573            ]1574        expect_less = [1575            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1576            mock.call.getHostVLUNs(self.FAKE_HOST),1577            mock.call.deleteVLUN(1578                self.VOLUME_3PAR_NAME,1579                None,1580                self.FAKE_HOST),1581            mock.call.getHostVLUNs(self.FAKE_HOST),1582            mock.call.logout()]1583        conn_info = self.driver.terminate_connection(self.volume,1584                                                     self.connector)1585        mock_client.assert_has_calls(expect_less)1586        self.assertNotIn('initiator_target_map', conn_info['data'])1587    def test_get_volume_stats(self):1588        # setup_mock_client drive with default configuration1589        # and return the mock HTTP 3PAR client1590        mock_client = self.setup_driver()1591        mock_client.getCPG.return_value = self.cpgs[0]1592        mock_client.getStorageSystemInfo.return_value = {'serialNumber':1593                                                         '1234'}1594        stats = self.driver.get_volume_stats(True)1595        self.assertEqual(stats['storage_protocol'], 'FC')1596        self.assertEqual(stats['total_capacity_gb'], 'infinite')1597        self.assertEqual(stats['free_capacity_gb'], 'infinite')1598        expected = [1599            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1600            mock.call.getCPG(HP3PAR_CPG),1601            mock.call.getStorageSystemInfo(),1602            mock.call.logout()]1603        mock_client.assert_has_calls(expected)1604        stats = self.driver.get_volume_stats(True)1605        self.assertEqual(stats['storage_protocol'], 'FC')1606        self.assertEqual(stats['total_capacity_gb'], 'infinite')1607        self.assertEqual(stats['free_capacity_gb'], 'infinite')1608        cpg2 = self.cpgs[0].copy()1609        cpg2.update({'SDGrowth': {'limitMiB': 8192}})1610        mock_client.getCPG.return_value = cpg21611        const = 0.00097656251612        stats = self.driver.get_volume_stats(True)1613        self.assertEqual(stats['storage_protocol'], 'FC')1614        total_capacity_gb = 8192 * const1615        self.assertEqual(stats['total_capacity_gb'], total_capacity_gb)1616        free_capacity_gb = int(1617            (8192 - self.cpgs[0]['UsrUsage']['usedMiB']) * const)1618        self.assertEqual(stats['free_capacity_gb'], free_capacity_gb)1619        self.driver.common.client.deleteCPG(HP3PAR_CPG)1620        self.driver.common.client.createCPG(HP3PAR_CPG, {})1621    def test_create_host(self):1622        # setup_mock_client drive with default configuration1623        # and return the mock HTTP 3PAR client1624        mock_client = self.setup_driver()1625        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1626        mock_client.getCPG.return_value = {}1627        mock_client.getHost.side_effect = [1628            hpexceptions.HTTPNotFound('fake'),1629            {'name': self.FAKE_HOST,1630                'FCPaths': [{'driverVersion': None,1631                             'firmwareVersion': None,1632                             'hostSpeed': 0,1633                             'model': None,1634                             'portPos': {'cardPort': 1, 'node': 1,1635                                         'slot': 2},1636                             'vendor': None,1637                             'wwn': self.wwn[0]},1638                            {'driverVersion': None,1639                             'firmwareVersion': None,1640                             'hostSpeed': 0,1641                             'model': None,1642                             'portPos': {'cardPort': 1, 'node': 0,1643                                         'slot': 2},1644                             'vendor': None,1645                             'wwn': self.wwn[1]}]}]1646        mock_client.findHost.return_value = None1647        mock_client.getVLUN.return_value = {'lun': 186}1648        host = self.driver._create_host(self.volume, self.connector)1649        expected = [1650            mock.call.getVolume('osv-0DM4qZEVSKON-DXN-NwVpw'),1651            mock.call.getCPG(HP3PAR_CPG),1652            mock.call.getHost(self.FAKE_HOST),1653            mock.call.findHost(wwn='123456789012345'),1654            mock.call.findHost(wwn='123456789054321'),1655            mock.call.createHost(1656                self.FAKE_HOST,1657                FCWwns=['123456789012345', '123456789054321'],1658                optional={'domain': None, 'persona': 1}),1659            mock.call.getHost(self.FAKE_HOST)]1660        mock_client.assert_has_calls(expected)1661        self.assertEqual(host['name'], self.FAKE_HOST)1662    def test_create_invalid_host(self):1663        # setup_mock_client drive with default configuration1664        # and return the mock HTTP 3PAR client1665        mock_client = self.setup_driver()1666        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1667        mock_client.getCPG.return_value = {}1668        mock_client.getHost.side_effect = [1669            hpexceptions.HTTPNotFound('Host not found.'), {1670                'name': 'fakehost.foo',1671                'FCPaths': [{'wwn': '123456789012345'}, {1672                    'wwn': '123456789054321'}]}]1673        mock_client.findHost.return_value = 'fakehost.foo'1674        host = self.driver._create_host(self.volume, self.connector)1675        expected = [1676            mock.call.getVolume('osv-0DM4qZEVSKON-DXN-NwVpw'),1677            mock.call.getCPG(HP3PAR_CPG),1678            mock.call.getHost('fakehost'),1679            mock.call.findHost(wwn='123456789012345'),1680            mock.call.getHost('fakehost.foo')]1681        mock_client.assert_has_calls(expected)1682        self.assertEqual(host['name'], 'fakehost.foo')1683    def test_create_modify_host(self):1684        # setup_mock_client drive with default configuration1685        # and return the mock HTTP 3PAR client1686        mock_client = self.setup_driver()1687        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1688        mock_client.getCPG.return_value = {}1689        mock_client.getHost.side_effect = [{1690            'name': self.FAKE_HOST, 'FCPaths': []},1691            {'name': self.FAKE_HOST,1692                'FCPaths': [{'wwn': '123456789012345'}, {1693                    'wwn': '123456789054321'}]}]1694        host = self.driver._create_host(self.volume, self.connector)1695        expected = [1696            mock.call.getVolume('osv-0DM4qZEVSKON-DXN-NwVpw'),1697            mock.call.getCPG(HP3PAR_CPG),1698            mock.call.getHost('fakehost'),1699            mock.call.modifyHost(1700                'fakehost', {1701                    'FCWWNs': ['123456789012345', '123456789054321'],1702                    'pathOperation': 1}),1703            mock.call.getHost('fakehost')]1704        mock_client.assert_has_calls(expected)1705        self.assertEqual(host['name'], self.FAKE_HOST)1706        self.assertEqual(len(host['FCPaths']), 2)1707    def test_modify_host_with_new_wwn(self):1708        # setup_mock_client drive with default configuration1709        # and return the mock HTTP 3PAR client1710        mock_client = self.setup_driver()1711        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1712        mock_client.getCPG.return_value = {}1713        getHost_ret1 = {1714            'name': self.FAKE_HOST,1715            'FCPaths': [{'wwn': '123456789054321'}]}1716        getHost_ret2 = {1717            'name': self.FAKE_HOST,1718            'FCPaths': [{'wwn': '123456789012345'},1719                        {'wwn': '123456789054321'}]}1720        mock_client.getHost.side_effect = [getHost_ret1, getHost_ret2]1721        host = self.driver._create_host(self.volume, self.connector)1722        expected = [1723            mock.call.getVolume('osv-0DM4qZEVSKON-DXN-NwVpw'),1724            mock.call.getCPG(HP3PAR_CPG),1725            mock.call.getHost('fakehost'),1726            mock.call.modifyHost(1727                'fakehost', {1728                    'FCWWNs': ['123456789012345'], 'pathOperation': 1}),1729            mock.call.getHost('fakehost')]1730        mock_client.assert_has_calls(expected)1731        self.assertEqual(host['name'], self.FAKE_HOST)1732        self.assertEqual(len(host['FCPaths']), 2)1733    def test_modify_host_with_unknown_wwn_and_new_wwn(self):1734        # setup_mock_client drive with default configuration1735        # and return the mock HTTP 3PAR client1736        mock_client = self.setup_driver()1737        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1738        mock_client.getCPG.return_value = {}1739        getHost_ret1 = {1740            'name': self.FAKE_HOST,1741            'FCPaths': [{'wwn': '123456789054321'},1742                        {'wwn': 'xxxxxxxxxxxxxxx'}]}1743        getHost_ret2 = {1744            'name': self.FAKE_HOST,1745            'FCPaths': [{'wwn': '123456789012345'},1746                        {'wwn': '123456789054321'},1747                        {'wwn': 'xxxxxxxxxxxxxxx'}]}1748        mock_client.getHost.side_effect = [getHost_ret1, getHost_ret2]1749        host = self.driver._create_host(self.volume, self.connector)1750        expected = [1751            mock.call.getVolume('osv-0DM4qZEVSKON-DXN-NwVpw'),1752            mock.call.getCPG(HP3PAR_CPG),1753            mock.call.getHost('fakehost'),1754            mock.call.modifyHost(1755                'fakehost', {1756                    'FCWWNs': ['123456789012345'], 'pathOperation': 1}),1757            mock.call.getHost('fakehost')]1758        mock_client.assert_has_calls(expected)1759        self.assertEqual(host['name'], self.FAKE_HOST)1760        self.assertEqual(len(host['FCPaths']), 3)1761class TestHP3PARISCSIDriver(HP3PARBaseDriver, test.TestCase):1762    TARGET_IQN = 'iqn.2000-05.com.3pardata:21810002ac00383d'1763    TARGET_LUN = 1861764    properties = {1765        'driver_volume_type': 'iscsi',1766        'data':1767        {'target_discovered': True,1768            'target_iqn': TARGET_IQN,1769            'target_lun': TARGET_LUN,1770            'target_portal': '1.1.1.2:1234'}}1771    def setup_driver(self, config=None, mock_conf=None):1772        self.ctxt = context.get_admin_context()1773        mock_client = self.setup_mock_client(1774            conf=config,1775            m_conf=mock_conf,1776            driver=hpdriver.HP3PARISCSIDriver)1777        expected = [1778            mock.call.setSSHOptions(1779                HP3PAR_SAN_IP,1780                HP3PAR_USER_NAME,1781                HP3PAR_USER_PASS,1782                privatekey=HP3PAR_SAN_SSH_PRIVATE,1783                port=HP3PAR_SAN_SSH_PORT,1784                conn_timeout=HP3PAR_SAN_SSH_CON_TIMEOUT),1785            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1786            mock.call.getCPG(HP3PAR_CPG),1787            mock.call.logout(),1788            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1789            mock.call.getPorts(),1790            mock.call.logout()]1791        mock_client.assert_has_calls(expected)1792        mock_client.reset_mock()1793        return mock_client1794    def test_initialize_connection(self):1795        # setup_mock_client drive with default configuration1796        # and return the mock HTTP 3PAR client1797        mock_client = self.setup_driver()1798        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1799        mock_client.getCPG.return_value = {}1800        mock_client.getHost.side_effect = [1801            hpexceptions.HTTPNotFound('fake'),1802            {'name': self.FAKE_HOST}]1803        mock_client.findHost.return_value = self.FAKE_HOST1804        mock_client.getHostVLUNs.return_value = [1805            {'active': True,1806             'volumeName': self.VOLUME_3PAR_NAME,1807             'lun': self.TARGET_LUN, 'type': 0}]1808        location = ("%(volume_name)s,%(lun_id)s,%(host)s,%(nsp)s" %1809                    {'volume_name': self.VOLUME_3PAR_NAME,1810                     'lun_id': self.TARGET_LUN,1811                     'host': self.FAKE_HOST,1812                     'nsp': 'something'})1813        mock_client.createVLUN.return_value = location1814        result = self.driver.initialize_connection(self.volume, self.connector)1815        expected = [1816            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1817            mock.call.getVolume(self.VOLUME_3PAR_NAME),1818            mock.call.getCPG(HP3PAR_CPG),1819            mock.call.getHost(self.FAKE_HOST),1820            mock.call.findHost(iqn='iqn.1993-08.org.debian:01:222'),1821            mock.call.getHost(self.FAKE_HOST),1822            mock.call.createVLUN(1823                self.VOLUME_3PAR_NAME,1824                auto=True,1825                hostname='fakehost',1826                portPos={'node': 8, 'slot': 1, 'cardPort': 1}),1827            mock.call.getHostVLUNs(self.FAKE_HOST),1828            mock.call.logout()]1829        mock_client.assert_has_calls(expected)1830        self.assertDictMatch(result, self.properties)1831    def test_get_volume_stats(self):1832        # setup_mock_client drive with default configuration1833        # and return the mock HTTP 3PAR client1834        mock_client = self.setup_driver()1835        mock_client.getCPG.return_value = self.cpgs[0]1836        mock_client.getStorageSystemInfo.return_value = {'serialNumber':1837                                                         '1234'}1838        stats = self.driver.get_volume_stats(True)1839        self.assertEqual(stats['storage_protocol'], 'iSCSI')1840        self.assertEqual(stats['total_capacity_gb'], 'infinite')1841        self.assertEqual(stats['free_capacity_gb'], 'infinite')1842        expected = [1843            mock.call.login(HP3PAR_USER_NAME, HP3PAR_USER_PASS),1844            mock.call.getCPG(HP3PAR_CPG),1845            mock.call.getStorageSystemInfo(),1846            mock.call.logout()]1847        mock_client.assert_has_calls(expected)1848        self.assertEqual(stats['storage_protocol'], 'iSCSI')1849        self.assertEqual(stats['total_capacity_gb'], 'infinite')1850        self.assertEqual(stats['free_capacity_gb'], 'infinite')1851        cpg2 = self.cpgs[0].copy()1852        cpg2.update({'SDGrowth': {'limitMiB': 8192}})1853        mock_client.getCPG.return_value = cpg21854        const = 0.00097656251855        stats = self.driver.get_volume_stats(True)1856        self.assertEqual(stats['storage_protocol'], 'iSCSI')1857        total_capacity_gb = 8192 * const1858        self.assertEqual(stats['total_capacity_gb'], total_capacity_gb)1859        free_capacity_gb = int(1860            (8192 - self.cpgs[0]['UsrUsage']['usedMiB']) * const)1861        self.assertEqual(stats['free_capacity_gb'], free_capacity_gb)1862    def test_create_host(self):1863        # setup_mock_client drive with default configuration1864        # and return the mock HTTP 3PAR client1865        mock_client = self.setup_driver()1866        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1867        mock_client.getCPG.return_value = {}1868        mock_client.getHost.side_effect = [1869            hpexceptions.HTTPNotFound('fake'),1870            {'name': self.FAKE_HOST}]1871        mock_client.findHost.return_value = None1872        mock_client.getVLUN.return_value = {'lun': self.TARGET_LUN}1873        host = self.driver._create_host(self.volume, self.connector)1874        expected = [1875            mock.call.getVolume('osv-0DM4qZEVSKON-DXN-NwVpw'),1876            mock.call.getCPG(HP3PAR_CPG),1877            mock.call.getHost(self.FAKE_HOST),1878            mock.call.findHost(iqn='iqn.1993-08.org.debian:01:222'),1879            mock.call.createHost(1880                self.FAKE_HOST,1881                optional={'domain': None, 'persona': 1},1882                iscsiNames=['iqn.1993-08.org.debian:01:222']),1883            mock.call.getHost(self.FAKE_HOST)]1884        mock_client.assert_has_calls(expected)1885        self.assertEqual(host['name'], self.FAKE_HOST)1886    def test_create_invalid_host(self):1887        # setup_mock_client drive with default configuration1888        # and return the mock HTTP 3PAR client1889        mock_client = self.setup_driver()1890        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1891        mock_client.getCPG.return_value = {}1892        mock_client.getHost.side_effect = [1893            hpexceptions.HTTPNotFound('Host not found.'),1894            {'name': 'fakehost.foo'}]1895        mock_client.findHost.return_value = 'fakehost.foo'1896        host = self.driver._create_host(self.volume, self.connector)1897        expected = [1898            mock.call.getVolume('osv-0DM4qZEVSKON-DXN-NwVpw'),1899            mock.call.getCPG(HP3PAR_CPG),1900            mock.call.getHost(self.FAKE_HOST),1901            mock.call.findHost(iqn='iqn.1993-08.org.debian:01:222'),1902            mock.call.getHost('fakehost.foo')]1903        mock_client.assert_has_calls(expected)1904        self.assertEqual(host['name'], 'fakehost.foo')1905    def test_create_modify_host(self):1906        # setup_mock_client drive with default configuration1907        # and return the mock HTTP 3PAR client1908        mock_client = self.setup_driver()1909        mock_client.getVolume.return_value = {'userCPG': HP3PAR_CPG}1910        mock_client.getCPG.return_value = {}1911        mock_client.getHost.side_effect = [1912            {'name': self.FAKE_HOST, 'FCPaths': []},1913            {'name': self.FAKE_HOST,1914             'FCPaths': [{'wwn': '123456789012345'},1915                         {'wwn': '123456789054321'}]}]1916        host = self.driver._create_host(self.volume, self.connector)1917        expected = [1918            mock.call.getVolume('osv-0DM4qZEVSKON-DXN-NwVpw'),1919            mock.call.getCPG(HP3PAR_CPG),1920            mock.call.getHost(self.FAKE_HOST),1921            mock.call.modifyHost(1922                self.FAKE_HOST,1923                {'pathOperation': 1,1924                    'iSCSINames': ['iqn.1993-08.org.debian:01:222']}),1925            mock.call.getHost(self.FAKE_HOST)]1926        mock_client.assert_has_calls(expected)1927        self.assertEqual(host['name'], self.FAKE_HOST)1928        self.assertEqual(len(host['FCPaths']), 2)1929    def test_get_least_used_nsp_for_host_single(self):1930        # setup_mock_client drive with default configuration1931        # and return the mock HTTP 3PAR client1932        mock_client = self.setup_driver()1933        mock_client.getPorts.return_value = PORTS_RET1934        mock_client.getVLUNs.return_value = VLUNS1_RET1935        #Setup a single ISCSI IP1936        iscsi_ips = ["10.10.220.253"]1937        self.driver.configuration.hp3par_iscsi_ips = iscsi_ips1938        self.driver.initialize_iscsi_ports()1939        nsp = self.driver._get_least_used_nsp_for_host('newhost')1940        self.assertEqual(nsp, "1:8:1")1941    def test_get_least_used_nsp_for_host_new(self):1942        # setup_mock_client drive with default configuration1943        # and return the mock HTTP 3PAR client1944        mock_client = self.setup_driver()1945        mock_client.getPorts.return_value = PORTS_RET1946        mock_client.getVLUNs.return_value = VLUNS1_RET1947        #Setup two ISCSI IPs1948        iscsi_ips = ["10.10.220.252", "10.10.220.253"]1949        self.driver.configuration.hp3par_iscsi_ips = iscsi_ips1950        self.driver.initialize_iscsi_ports()1951        # Host 'newhost' does not yet have any iscsi paths,1952        # so the 'least used' is returned1953        nsp = self.driver._get_least_used_nsp_for_host('newhost')1954        self.assertEqual(nsp, "1:8:2")1955    def test_get_least_used_nsp_for_host_reuse(self):1956        # setup_mock_client drive with default configuration1957        # and return the mock HTTP 3PAR client1958        mock_client = self.setup_driver()1959        mock_client.getPorts.return_value = PORTS_RET1960        mock_client.getVLUNs.return_value = VLUNS1_RET1961        #Setup two ISCSI IPs1962        iscsi_ips = ["10.10.220.252", "10.10.220.253"]1963        self.driver.configuration.hp3par_iscsi_ips = iscsi_ips1964        self.driver.initialize_iscsi_ports()1965        # hosts 'foo' and 'bar' already have active iscsi paths1966        # the same one should be used1967        nsp = self.driver._get_least_used_nsp_for_host('foo')1968        self.assertEqual(nsp, "1:8:2")1969        nsp = self.driver._get_least_used_nsp_for_host('bar')1970        self.assertEqual(nsp, "1:8:1")1971    def test_get_least_used_nps_for_host_fc(self):1972        # setup_mock_client drive with default configuration1973        # and return the mock HTTP 3PAR client1974        mock_client = self.setup_driver()1975        mock_client.getPorts.return_value = PORTS1_RET1976        mock_client.getVLUNs.return_value = VLUNS5_RET1977        #Setup two ISCSI IPs1978        iscsi_ips = ["10.10.220.252", "10.10.220.253"]1979        self.driver.configuration.hp3par_iscsi_ips = iscsi_ips1980        self.driver.initialize_iscsi_ports()1981        nsp = self.driver._get_least_used_nsp_for_host('newhost')1982        self.assertNotEqual(nsp, "0:6:3")1983        self.assertEqual(nsp, "1:8:1")1984    def test_invalid_iscsi_ip(self):1985        config = self.setup_configuration()1986        config.hp3par_iscsi_ips = ['10.10.220.250', '10.10.220.251']1987        config.iscsi_ip_address = '10.10.10.10'1988        mock_conf = {1989            'getPorts.return_value': {1990                'members': [1991                    {'portPos': {'node': 1, 'slot': 8, 'cardPort': 2},1992                     'protocol': 2,1993                     'IPAddr': '10.10.220.252',1994                     'linkState': 4,1995                     'device': [],1996                     'iSCSIName': self.TARGET_IQN,1997                     'mode': 2,1998                     'HWAddr': '2C27D75375D2',1999                     'type': 8},2000                    {'portPos': {'node': 1, 'slot': 8, 'cardPort': 1},2001                     'protocol': 2,2002                     'IPAddr': '10.10.220.253',2003                     'linkState': 4,2004                     'device': [],2005                     'iSCSIName': self.TARGET_IQN,2006                     'mode': 2,2007                     'HWAddr': '2C27D75375D6',2008                     'type': 8}]}}2009        # no valid ip addr should be configured.2010        self.assertRaises(exception.InvalidInput,2011                          self.setup_driver,2012                          config=config,2013                          mock_conf=mock_conf)2014    def test_get_least_used_nsp(self):2015        # setup_mock_client drive with default configuration2016        # and return the mock HTTP 3PAR client2017        mock_client = self.setup_driver()2018        ports = [2019            {'portPos': {'node': 1, 'slot': 8, 'cardPort': 2}, 'active': True},2020            {'portPos': {'node': 1, 'slot': 8, 'cardPort': 1}, 'active': True},2021            {'portPos': {'node': 1, 'slot': 8, 'cardPort': 2}, 'active': True},2022            {'portPos': {'node': 0, 'slot': 2, 'cardPort': 2}, 'active': True},2023            {'portPos': {'node': 0, 'slot': 2, 'cardPort': 1}, 'active': True},2024            {'portPos': {'node': 0, 'slot': 2, 'cardPort': 1}, 'active': True},2025            {'portPos': {'node': 0, 'slot': 2, 'cardPort': 1}, 'active': True},2026            {'portPos': {'node': 0, 'slot': 2, 'cardPort': 1}, 'active': True}]2027        mock_client.getVLUNs.return_value = {'members': ports}2028        # in use count2029        vluns = self.driver.common.client.getVLUNs()2030        nsp = self.driver._get_least_used_nsp(vluns['members'],2031                                              ['0:2:1', '1:8:1'])...

test_xio.py

Source:test_xio.py

...661        elif self.protocol == 'fibre_channel':662            self.configuration.ise_protocol = protocol663            self.connector = FC_CONN1664            self.hostgid = self.connector['wwpns'][0]665    def setup_driver(self):666        # this setups up driver object with previously set configuration values667        if self.configuration.ise_protocol == 'iscsi':668            self.driver =\669                xio.XIOISEISCSIDriver(configuration=self.configuration)670        elif self.configuration.ise_protocol == 'fibre_channel':671            self.driver =\672                xio.XIOISEFCDriver(configuration=self.configuration)673        elif self.configuration.ise_protocol == 'test_prot':674            # if test_prot specified override with correct protocol675            # used to bypass protocol specific driver676            self.configuration.ise_protocol = self.protocol677            self.driver = xio.XIOISEDriver(configuration=self.configuration)678        else:679            # Invalid protocol type680            raise exception.Invalid()681#################################682#         UNIT TESTS            #683#################################684    def test_do_setup(self, mock_req):685        self.setup_driver()686        mock_req.side_effect = iter([ISE_GET_QUERY_RESP])687        self.driver.do_setup(None)688    def test_negative_do_setup_no_clone_support(self, mock_req):689        self.setup_driver()690        mock_req.side_effect = iter([ISE_GET_QUERY_NO_CLONE_RESP])691        self.assertRaises(exception.XIODriverException,692                          self.driver.do_setup, None)693    def test_negative_do_setup_no_capabilities(self, mock_req):694        self.setup_driver()695        mock_req.side_effect = iter([ISE_GET_QUERY_NO_CAP_RESP])696        self.assertRaises(exception.XIODriverException,697                          self.driver.do_setup, None)698    def test_negative_do_setup_no_ctrl(self, mock_req):699        self.setup_driver()700        mock_req.side_effect = iter([ISE_GET_QUERY_NO_CTRL_RESP])701        self.assertRaises(exception.XIODriverException,702                          self.driver.do_setup, None)703    def test_negative_do_setup_no_ipaddress(self, mock_req):704        self.setup_driver()705        mock_req.side_effect = iter([ISE_GET_QUERY_NO_IP_RESP])706        self.driver.do_setup(None)707    def test_negative_do_setup_bad_globalid_none(self, mock_req):708        self.setup_driver()709        mock_req.side_effect = iter([ISE_GET_QUERY_NO_GID_RESP])710        self.assertRaises(exception.XIODriverException,711                          self.driver.do_setup, None)712    def test_check_for_setup_error(self, mock_req):713        mock_req.side_effect = iter([ISE_GET_QUERY_RESP])714        self.setup_driver()715        self.driver.check_for_setup_error()716    def test_negative_do_setup_bad_ip(self, mock_req):717        # set san_ip to bad value718        self.configuration.san_ip = ''719        mock_req.side_effect = iter([ISE_GET_QUERY_RESP])720        self.setup_driver()721        self.assertRaises(exception.XIODriverException,722                          self.driver.check_for_setup_error)723    def test_negative_do_setup_bad_user_blank(self, mock_req):724        # set san_user to bad value725        self.configuration.san_login = ''726        mock_req.side_effect = iter([ISE_GET_QUERY_RESP])727        self.setup_driver()728        self.assertRaises(exception.XIODriverException,729                          self.driver.check_for_setup_error)730    def test_negative_do_setup_bad_password_blank(self, mock_req):731        # set san_password to bad value732        self.configuration.san_password = ''733        mock_req.side_effect = iter([ISE_GET_QUERY_RESP])734        self.setup_driver()735        self.assertRaises(exception.XIODriverException,736                          self.driver.check_for_setup_error)737    def test_get_volume_stats(self, mock_req):738        self.setup_driver()739        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,740                                     ISE_GET_STORAGE_POOLS_RESP])741        backend_name = self.configuration.volume_backend_name742        if self.configuration.ise_protocol == 'iscsi':743            protocol = 'iSCSI'744        else:745            protocol = 'fibre_channel'746        exp_result = {'vendor_name': "X-IO",747                      'driver_version': "1.1.4",748                      'volume_backend_name': backend_name,749                      'reserved_percentage': 0,750                      'total_capacity_gb': 100,751                      'free_capacity_gb': 60,752                      'QoS_support': True,753                      'affinity': True,754                      'thin': False,755                      'pools': [{'pool_ise_name': "Pool 1",756                                 'pool_name': "1",757                                 'status': "Operational",758                                 'status_details': "None",759                                 'free_capacity_gb': 60,760                                 'free_capacity_gb_raid_0': 60,761                                 'free_capacity_gb_raid_1': 30,762                                 'free_capacity_gb_raid_5': 45,763                                 'allocated_capacity_gb': 40,764                                 'allocated_capacity_gb_raid_0': 0,765                                 'allocated_capacity_gb_raid_1': 40,766                                 'allocated_capacity_gb_raid_5': 0,767                                 'health': 100,768                                 'media': "Hybrid",769                                 'total_capacity_gb': 100,770                                 'QoS_support': True,771                                 'reserved_percentage': 0}],772                      'active_volumes': 2,773                      'storage_protocol': protocol}774        act_result = self.driver.get_volume_stats(True)775        self.assertDictEqual(exp_result, act_result)776    def test_get_volume_stats_ssl(self, mock_req):777        self.configuration.driver_use_ssl = True778        self.setup_driver()779        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,780                                     ISE_GET_STORAGE_POOLS_RESP])781        self.driver.get_volume_stats(True)782    def test_negative_get_volume_stats_bad_primary(self, mock_req):783        self.configuration.ise_connection_retries = 1784        self.setup_driver()785        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,786                                     ISE_BAD_CONNECTION_RESP,787                                     ISE_GET_STORAGE_POOLS_RESP])788        self.driver.get_volume_stats(True)789    def test_create_volume(self, mock_req):790        ctxt = context.get_admin_context()791        extra_specs = {"Feature:Pool": "1",792                       "Feature:Raid": "1",793                       "Affinity:Type": "flash",794                       "Alloc:Type": "thick"}795        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)796        specs = {'qos:minIOPS': '20',797                 'qos:maxIOPS': '2000',798                 'qos:burstIOPS': '5000'}799        qos = qos_specs.create(ctxt, 'fake-qos', specs)800        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])801        VOLUME1['volume_type_id'] = type_ref['id']802        self.setup_driver()803        if self.configuration.ise_protocol == 'iscsi':804            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,805                                         ISE_CREATE_VOLUME_RESP,806                                         ISE_GET_VOL1_STATUS_RESP,807                                         ISE_GET_IONETWORKS_RESP])808            exp_result = {}809            exp_result = {"provider_auth": ""}810            act_result = self.driver.create_volume(VOLUME1)811            self.assertDictEqual(exp_result, act_result)812        elif self.configuration.ise_protocol == 'fibre_channel':813            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,814                                         ISE_CREATE_VOLUME_RESP,815                                         ISE_GET_VOL1_STATUS_RESP])816            self.driver.create_volume(VOLUME1)817    def test_create_volume_chap(self, mock_req):818        ctxt = context.get_admin_context()819        extra_specs = {"Feature:Pool": "1",820                       "Feature:Raid": "1",821                       "Affinity:Type": "flash",822                       "Alloc:Type": "thick"}823        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)824        specs = {'qos:minIOPS': '20',825                 'qos:maxIOPS': '2000',826                 'qos:burstIOPS': '5000'}827        qos = qos_specs.create(ctxt, 'fake-qos', specs)828        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])829        VOLUME1['volume_type_id'] = type_ref['id']830        self.setup_driver()831        if self.configuration.ise_protocol == 'iscsi':832            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,833                                         ISE_CREATE_VOLUME_RESP,834                                         ISE_GET_VOL1_STATUS_RESP,835                                         ISE_GET_IONETWORKS_CHAP_RESP])836            exp_result = {}837            exp_result = {"provider_auth": "CHAP abc abc"}838            act_result = self.driver.create_volume(VOLUME1)839            self.assertDictEqual(exp_result, act_result)840        elif self.configuration.ise_protocol == 'fibre_channel':841            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,842                                         ISE_CREATE_VOLUME_RESP,843                                         ISE_GET_VOL1_STATUS_RESP])844            self.driver.create_volume(VOLUME1)845    def test_create_volume_type_none(self, mock_req):846        self.setup_driver()847        if self.configuration.ise_protocol == 'iscsi':848            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,849                                         ISE_CREATE_VOLUME_RESP,850                                         ISE_GET_VOL1_STATUS_RESP,851                                         ISE_GET_IONETWORKS_RESP])852        elif self.configuration.ise_protocol == 'fibre_channel':853            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,854                                         ISE_CREATE_VOLUME_RESP,855                                         ISE_GET_VOL1_STATUS_RESP])856        self.driver.create_volume(VOLUME3)857    def test_delete_volume(self, mock_req):858        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,859                                     ISE_GET_ALLOC_WITH_EP_RESP,860                                     ISE_DELETE_ALLOC_RESP,861                                     ISE_GET_VOL1_STATUS_RESP,862                                     ISE_DELETE_VOLUME_RESP,863                                     ISE_GET_VOL_STATUS_404_RESP])864        self.setup_driver()865        self.driver.delete_volume(VOLUME1)866    def test_delete_volume_delayed(self, mock_req):867        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,868                                     ISE_GET_ALLOC_WITH_EP_RESP,869                                     ISE_DELETE_ALLOC_RESP,870                                     ISE_GET_VOL1_STATUS_RESP,871                                     ISE_DELETE_VOLUME_RESP,872                                     ISE_GET_VOL1_STATUS_RESP,873                                     ISE_GET_VOL1_STATUS_RESP,874                                     ISE_GET_VOL_STATUS_404_RESP])875        self.setup_driver()876        self.driver.delete_volume(VOLUME1)877    def test_delete_volume_timeout(self, mock_req):878        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,879                                     ISE_GET_ALLOC_WITH_EP_RESP,880                                     ISE_DELETE_ALLOC_RESP,881                                     ISE_GET_VOL1_STATUS_RESP,882                                     ISE_DELETE_VOLUME_RESP,883                                     ISE_GET_VOL1_STATUS_RESP,884                                     ISE_GET_VOL1_STATUS_RESP,885                                     ISE_GET_VOL1_STATUS_RESP])886        self.configuration.ise_completion_retries = 3887        self.setup_driver()888        self.driver.delete_volume(VOLUME1)889    def test_delete_volume_none_existing(self, mock_req):890        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,891                                     ISE_GET_ALLOC_WITH_EP_RESP,892                                     ISE_DELETE_ALLOC_RESP,893                                     ISE_GET_VOL1_STATUS_RESP])894        self.setup_driver()895        self.driver.delete_volume(VOLUME2)896    def test_initialize_connection_positive(self, mock_req):897        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,898                                     ISE_GET_HOSTS_HOST2_RESP,899                                     ISE_CREATE_HOST_RESP,900                                     ISE_GET_HOSTS_HOST1_RESP,901                                     ISE_CREATE_ALLOC_RESP,902                                     ISE_GET_ALLOC_WITH_EP_RESP,903                                     ISE_GET_CONTROLLERS_RESP])904        self.setup_driver()905        exp_result = {}906        if self.configuration.ise_protocol == 'iscsi':907            exp_result = {"driver_volume_type": "iscsi",908                          "data": {"target_lun": 1,909                                   "volume_id": '1',910                                   "target_discovered": False,911                                   "target_iqn": ISE_IQN,912                                   "target_portal": ISE_ISCSI_IP1 + ":3260"}}913        elif self.configuration.ise_protocol == 'fibre_channel':914            exp_result = {"driver_volume_type": "fibre_channel",915                          "data": {"target_lun": 1,916                                   "volume_id": '1',917                                   "target_discovered": True,918                                   "initiator_target_map": ISE_INIT_TARGET_MAP,919                                   "target_wwn": ISE_TARGETS}}920        act_result =\921            self.driver.initialize_connection(VOLUME1, self.connector)922        self.assertDictEqual(exp_result, act_result)923    def test_initialize_connection_positive_host_type(self, mock_req):924        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,925                                     ISE_GET_HOSTS_HOST1_HOST_TYPE_RESP,926                                     ISE_MODIFY_HOST_RESP,927                                     ISE_CREATE_ALLOC_RESP,928                                     ISE_GET_ALLOC_WITH_EP_RESP,929                                     ISE_GET_CONTROLLERS_RESP])930        self.setup_driver()931        exp_result = {}932        if self.configuration.ise_protocol == 'iscsi':933            exp_result = {"driver_volume_type": "iscsi",934                          "data": {"target_lun": 1,935                                   "volume_id": '1',936                                   "target_discovered": False,937                                   "target_iqn": ISE_IQN,938                                   "target_portal": ISE_ISCSI_IP1 + ":3260"}}939        elif self.configuration.ise_protocol == 'fibre_channel':940            exp_result = {"driver_volume_type": "fibre_channel",941                          "data": {"target_lun": 1,942                                   "volume_id": '1',943                                   "target_discovered": True,944                                   "initiator_target_map": ISE_INIT_TARGET_MAP,945                                   "target_wwn": ISE_TARGETS}}946        act_result =\947            self.driver.initialize_connection(VOLUME1, self.connector)948        self.assertDictEqual(exp_result, act_result)949    def test_initialize_connection_positive_chap(self, mock_req):950        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,951                                     ISE_GET_HOSTS_HOST2_RESP,952                                     ISE_CREATE_HOST_RESP,953                                     ISE_GET_HOSTS_HOST1_RESP,954                                     ISE_CREATE_ALLOC_RESP,955                                     ISE_GET_ALLOC_WITH_EP_RESP,956                                     ISE_GET_CONTROLLERS_RESP])957        self.setup_driver()958        exp_result = {}959        if self.configuration.ise_protocol == 'iscsi':960            exp_result = {"driver_volume_type": "iscsi",961                          "data": {"target_lun": 1,962                                   "volume_id": '2',963                                   "target_discovered": False,964                                   "target_iqn": ISE_IQN,965                                   "target_portal": ISE_ISCSI_IP1 + ":3260",966                                   'auth_method': 'CHAP',967                                   'auth_username': 'abc',968                                   'auth_password': 'abc'}}969        elif self.configuration.ise_protocol == 'fibre_channel':970            exp_result = {"driver_volume_type": "fibre_channel",971                          "data": {"target_lun": 1,972                                   "volume_id": '2',973                                   "target_discovered": True,974                                   "initiator_target_map": ISE_INIT_TARGET_MAP,975                                   "target_wwn": ISE_TARGETS}}976        act_result =\977            self.driver.initialize_connection(VOLUME2, self.connector)978        self.assertDictEqual(exp_result, act_result)979    def test_initialize_connection_negative_no_host(self, mock_req):980        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,981                                     ISE_GET_HOSTS_HOST2_RESP,982                                     ISE_CREATE_HOST_RESP,983                                     ISE_GET_HOSTS_HOST2_RESP])984        self.setup_driver()985        self.assertRaises(exception.XIODriverException,986                          self.driver.initialize_connection,987                          VOLUME2, self.connector)988    def test_initialize_connection_negative_host_type(self, mock_req):989        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,990                                     ISE_GET_HOSTS_HOST1_HOST_TYPE_RESP,991                                     ISE_400_RESP])992        self.setup_driver()993        self.assertRaises(exception.XIODriverException,994                          self.driver.initialize_connection,995                          VOLUME2, self.connector)996    def test_terminate_connection_positive(self, mock_req):997        self.setup_driver()998        if self.configuration.ise_protocol == 'iscsi':999            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1000                                         ISE_GET_HOSTS_HOST1_RESP,1001                                         ISE_GET_ALLOC_WITH_EP_RESP,1002                                         ISE_DELETE_ALLOC_RESP,1003                                         ISE_GET_ALLOC_WITH_EP_RESP,1004                                         ISE_GET_HOSTS_HOST1_RESP,1005                                         ISE_DELETE_ALLOC_RESP])1006        elif self.configuration.ise_protocol == 'fibre_channel':1007            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1008                                         ISE_GET_HOSTS_HOST1_RESP,1009                                         ISE_GET_ALLOC_WITH_EP_RESP,1010                                         ISE_DELETE_ALLOC_RESP,1011                                         ISE_GET_ALLOC_WITH_EP_RESP,1012                                         ISE_GET_CONTROLLERS_RESP,1013                                         ISE_GET_HOSTS_HOST1_RESP,1014                                         ISE_DELETE_ALLOC_RESP])1015        self.driver.terminate_connection(VOLUME1, self.connector)1016    def test_terminate_connection_positive_noalloc(self, mock_req):1017        self.setup_driver()1018        if self.configuration.ise_protocol == 'iscsi':1019            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1020                                         ISE_GET_HOSTS_HOST1_RESP,1021                                         ISE_GET_ALLOC_WITH_NO_ALLOC_RESP,1022                                         ISE_GET_ALLOC_WITH_NO_ALLOC_RESP,1023                                         ISE_GET_HOSTS_HOST1_RESP,1024                                         ISE_DELETE_ALLOC_RESP])1025        elif self.configuration.ise_protocol == 'fibre_channel':1026            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1027                                         ISE_GET_HOSTS_HOST1_RESP,1028                                         ISE_GET_ALLOC_WITH_NO_ALLOC_RESP,1029                                         ISE_GET_ALLOC_WITH_NO_ALLOC_RESP,1030                                         ISE_GET_CONTROLLERS_RESP,1031                                         ISE_GET_HOSTS_HOST1_RESP,1032                                         ISE_DELETE_ALLOC_RESP])1033        self.driver.terminate_connection(VOLUME1, self.connector)1034    def test_negative_terminate_connection_bad_host(self, mock_req):1035        self.setup_driver()1036        test_connector = {}1037        if self.configuration.ise_protocol == 'iscsi':1038            test_connector['initiator'] = 'bad_iqn'1039            test_connector['host'] = ''1040            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1041                                         ISE_GET_HOSTS_HOST1_RESP])1042        elif self.configuration.ise_protocol == 'fibre_channel':1043            test_connector['wwpns'] = 'bad_wwn'1044            test_connector['host'] = ''1045            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1046                                         ISE_GET_HOSTS_HOST1_RESP,1047                                         ISE_GET_CONTROLLERS_RESP])1048        self.driver.terminate_connection(VOLUME1, test_connector)1049    def test_create_snapshot(self, mock_req):1050        ctxt = context.get_admin_context()1051        extra_specs = {"Feature:Pool": "1",1052                       "Feature:Raid": "1",1053                       "Affinity:Type": "flash",1054                       "Alloc:Type": "thick"}1055        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)1056        specs = {'qos:minIOPS': '20',1057                 'qos:maxIOPS': '2000',1058                 'qos:burstIOPS': '5000'}1059        qos = qos_specs.create(ctxt, 'fake-qos', specs)1060        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1061        SNAPSHOT1['volume_type_id'] = type_ref['id']1062        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1063                                     ISE_GET_VOL1_STATUS_RESP,1064                                     ISE_PREP_SNAPSHOT_RESP,1065                                     ISE_GET_SNAP1_STATUS_RESP,1066                                     ISE_CREATE_SNAPSHOT_RESP,1067                                     ISE_GET_SNAP1_STATUS_RESP])1068        self.setup_driver()1069        self.driver.create_snapshot(SNAPSHOT1)1070    @mock.patch('oslo_service.loopingcall.FixedIntervalLoopingCall',1071                new=utils.ZeroIntervalLoopingCall)1072    def test_negative_create_snapshot_invalid_state_recover(self, mock_req):1073        ctxt = context.get_admin_context()1074        extra_specs = {"Feature:Pool": "1",1075                       "Feature:Raid": "1",1076                       "Affinity:Type": "flash",1077                       "Alloc:Type": "thick"}1078        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)1079        specs = {'qos:minIOPS': '20',1080                 'qos:maxIOPS': '2000',1081                 'qos:burstIOPS': '5000'}1082        qos = qos_specs.create(ctxt, 'fake-qos', specs)1083        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1084        SNAPSHOT1['volume_type_id'] = type_ref['id']1085        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1086                                     ISE_GET_VOL1_STATUS_RESP,1087                                     ISE_400_INVALID_STATE_RESP,1088                                     ISE_PREP_SNAPSHOT_RESP,1089                                     ISE_GET_SNAP1_STATUS_RESP,1090                                     ISE_CREATE_SNAPSHOT_RESP,1091                                     ISE_GET_SNAP1_STATUS_RESP])1092        self.setup_driver()1093        self.driver.create_snapshot(SNAPSHOT1)1094    @mock.patch('oslo_service.loopingcall.FixedIntervalLoopingCall',1095                new=utils.ZeroIntervalLoopingCall)1096    def test_negative_create_snapshot_invalid_state_norecover(self, mock_req):1097        ctxt = context.get_admin_context()1098        extra_specs = {"Feature:Pool": "1",1099                       "Feature:Raid": "1",1100                       "Affinity:Type": "flash",1101                       "Alloc:Type": "thick"}1102        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)1103        specs = {'qos:minIOPS': '20',1104                 'qos:maxIOPS': '2000',1105                 'qos:burstIOPS': '5000'}1106        qos = qos_specs.create(ctxt, 'fake-qos', specs)1107        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1108        SNAPSHOT1['volume_type_id'] = type_ref['id']1109        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1110                                     ISE_GET_VOL1_STATUS_RESP,1111                                     ISE_400_INVALID_STATE_RESP,1112                                     ISE_400_INVALID_STATE_RESP,1113                                     ISE_400_INVALID_STATE_RESP,1114                                     ISE_400_INVALID_STATE_RESP,1115                                     ISE_400_INVALID_STATE_RESP])1116        self.configuration.ise_completion_retries = 51117        self.setup_driver()1118        self.assertRaises(exception.XIODriverException,1119                          self.driver.create_snapshot, SNAPSHOT1)1120    def test_negative_create_snapshot_conflict(self, mock_req):1121        ctxt = context.get_admin_context()1122        extra_specs = {"Feature:Pool": "1",1123                       "Feature:Raid": "1",1124                       "Affinity:Type": "flash",1125                       "Alloc:Type": "thick"}1126        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)1127        specs = {'qos:minIOPS': '20',1128                 'qos:maxIOPS': '2000',1129                 'qos:burstIOPS': '5000'}1130        qos = qos_specs.create(ctxt, 'fake-qos', specs)1131        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1132        SNAPSHOT1['volume_type_id'] = type_ref['id']1133        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1134                                     ISE_GET_VOL1_STATUS_RESP,1135                                     ISE_409_CONFLICT_RESP])1136        self.configuration.ise_completion_retries = 11137        self.setup_driver()1138        self.assertRaises(exception.XIODriverException,1139                          self.driver.create_snapshot, SNAPSHOT1)1140    def test_delete_snapshot(self, mock_req):1141        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1142                                     ISE_GET_ALLOC_WITH_EP_RESP,1143                                     ISE_DELETE_ALLOC_RESP,1144                                     ISE_GET_SNAP1_STATUS_RESP,1145                                     ISE_DELETE_VOLUME_RESP])1146        self.setup_driver()1147        self.driver.delete_snapshot(SNAPSHOT1)1148    def test_clone_volume(self, mock_req):1149        ctxt = context.get_admin_context()1150        extra_specs = {"Feature:Pool": "1",1151                       "Feature:Raid": "1",1152                       "Affinity:Type": "flash",1153                       "Alloc:Type": "thick"}1154        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)1155        specs = {'qos:minIOPS': '20',1156                 'qos:maxIOPS': '2000',1157                 'qos:burstIOPS': '5000'}1158        qos = qos_specs.create(ctxt, 'fake-qos', specs)1159        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1160        VOLUME1['volume_type_id'] = type_ref['id']1161        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1162                                     ISE_GET_VOL1_STATUS_RESP,1163                                     ISE_PREP_SNAPSHOT_RESP,1164                                     ISE_GET_SNAP1_STATUS_RESP,1165                                     ISE_CREATE_SNAPSHOT_RESP,1166                                     ISE_GET_SNAP1_STATUS_RESP])1167        self.setup_driver()1168        self.driver.create_cloned_volume(CLONE1, VOLUME1)1169    def test_extend_volume(self, mock_req):1170        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1171                                     ISE_GET_VOL1_STATUS_RESP,1172                                     ISE_MODIFY_VOLUME_RESP])1173        self.setup_driver()1174        self.driver.extend_volume(VOLUME1, NEW_VOLUME_SIZE)1175    def test_retype_volume(self, mock_req):1176        ctxt = context.get_admin_context()1177        extra_specs = {"Feature:Pool": "1",1178                       "Feature:Raid": "1",1179                       "Affinity:Type": "flash",1180                       "Alloc:Type": "thick"}1181        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)1182        specs = {'qos:minIOPS': '20',1183                 'qos:maxIOPS': '2000',1184                 'qos:burstIOPS': '5000'}1185        qos = qos_specs.create(ctxt, 'fake-qos', specs)1186        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1187        VOLUME1['volume_type_id'] = type_ref['id']1188        # New volume type1189        extra_specs = {"Feature:Pool": "1",1190                       "Feature:Raid": "5",1191                       "Affinity:Type": "flash",1192                       "Alloc:Type": "thick"}1193        type_ref = volume_types.create(ctxt, 'VT2', extra_specs)1194        specs = {'qos:minIOPS': '30',1195                 'qos:maxIOPS': '3000',1196                 'qos:burstIOPS': '10000'}1197        qos = qos_specs.create(ctxt, 'fake-qos2', specs)1198        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1199        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1200                                     ISE_GET_VOL1_STATUS_RESP,1201                                     ISE_MODIFY_VOLUME_RESP])1202        self.setup_driver()1203        self.driver.retype(ctxt, VOLUME1, type_ref, 0, 0)1204    def test_create_volume_from_snapshot(self, mock_req):1205        ctxt = context.get_admin_context()1206        extra_specs = {"Feature:Pool": "1",1207                       "Feature:Raid": "1",1208                       "Affinity:Type": "flash",1209                       "Alloc:Type": "thick"}1210        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)1211        specs = {'qos:minIOPS': '20',1212                 'qos:maxIOPS': '2000',1213                 'qos:burstIOPS': '5000'}1214        qos = qos_specs.create(ctxt, 'fake-qos', specs)1215        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1216        SNAPSHOT1['volume_type_id'] = type_ref['id']1217        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1218                                     ISE_GET_SNAP1_STATUS_RESP,1219                                     ISE_PREP_SNAPSHOT_RESP,1220                                     ISE_GET_VOL1_STATUS_RESP,1221                                     ISE_CREATE_SNAPSHOT_RESP,1222                                     ISE_GET_VOL1_STATUS_RESP])1223        self.setup_driver()1224        self.driver.create_volume_from_snapshot(VOLUME1, SNAPSHOT1)1225    def test_manage_existing(self, mock_req):1226        ctxt = context.get_admin_context()1227        extra_specs = {"Feature:Pool": "1",1228                       "Feature:Raid": "1",1229                       "Affinity:Type": "flash",1230                       "Alloc:Type": "thick"}1231        type_ref = volume_types.create(ctxt, 'VT1', extra_specs)1232        specs = {'qos:minIOPS': '20',1233                 'qos:maxIOPS': '2000',1234                 'qos:burstIOPS': '5000'}1235        qos = qos_specs.create(ctxt, 'fake-qos', specs)1236        qos_specs.associate_qos_with_type(ctxt, qos['id'], type_ref['id'])1237        VOLUME1['volume_type_id'] = type_ref['id']1238        self.setup_driver()1239        if self.configuration.ise_protocol == 'iscsi':1240            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1241                                         ISE_GET_VOL1_STATUS_RESP,1242                                         ISE_MODIFY_VOLUME_RESP,1243                                         ISE_GET_IONETWORKS_RESP])1244        elif self.configuration.ise_protocol == 'fibre_channel':1245            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1246                                         ISE_GET_VOL1_STATUS_RESP,1247                                         ISE_MODIFY_VOLUME_RESP])1248        self.driver.manage_existing(VOLUME1, {'source-name': 'testvol'})1249    def test_manage_existing_no_source_name(self, mock_req):1250        self.setup_driver()1251        if self.configuration.ise_protocol == 'iscsi':1252            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1253                                         ISE_GET_VOL1_STATUS_RESP,1254                                         ISE_MODIFY_VOLUME_RESP,1255                                         ISE_GET_IONETWORKS_RESP])1256        elif self.configuration.ise_protocol == 'fibre_channel':1257            mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1258                                         ISE_GET_VOL1_STATUS_RESP,1259                                         ISE_MODIFY_VOLUME_RESP])1260        self.assertRaises(exception.XIODriverException,1261                          self.driver.manage_existing, VOLUME1, {})1262    def test_manage_existing_get_size(self, mock_req):1263        self.setup_driver()1264        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1265                                     ISE_GET_VOL1_STATUS_RESP])1266        exp_result = 101267        act_result = \1268            self.driver.manage_existing_get_size(VOLUME1,1269                                                 {'source-name': 'a'})1270        self.assertEqual(exp_result, act_result)1271    def test_manage_existing_get_size_no_source_name(self, mock_req):1272        self.setup_driver()1273        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1274                                     ISE_GET_VOL1_STATUS_RESP])1275        self.assertRaises(exception.XIODriverException,1276                          self.driver.manage_existing_get_size, VOLUME1, {})1277    def test_unmanage(self, mock_req):1278        self.setup_driver()1279        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1280                                    ISE_GET_VOL1_STATUS_RESP])1281        self.driver.unmanage(VOLUME1)1282    def test_negative_unmanage_no_volume_status_xml(self, mock_req):1283        self.setup_driver()1284        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1285                                    ISE_GET_VOL_STATUS_NO_STATUS_RESP])1286        self.driver.unmanage(VOLUME1)1287    def test_negative_unmanage_no_volume_xml(self, mock_req):1288        self.setup_driver()1289        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1290                                    ISE_GET_VOL_STATUS_NO_VOL_NODE_RESP])1291        self.assertRaises(exception.XIODriverException,1292                          self.driver.unmanage, VOLUME1)1293    def test_negative_unmanage_non_existing_volume(self, mock_req):1294        self.setup_driver()1295        mock_req.side_effect = iter([ISE_GET_QUERY_RESP,1296                                     ISE_GET_VOL_STATUS_404_RESP])1297        self.assertRaises(exception.XIODriverException,1298                          self.driver.unmanage, VOLUME1)1299class XIOISEISCSIDriverTestCase(XIOISEDriverTestCase, test.TestCase):1300    def setUp(self):1301        super(XIOISEISCSIDriverTestCase, self).setUp()1302        self.setup_test('iscsi')1303class XIOISEFCDriverTestCase(XIOISEDriverTestCase, test.TestCase):1304    def setUp(self):1305        super(XIOISEFCDriverTestCase, self).setUp()...

test_main_page.py

Source:test_main_page.py

1from .pages.main_page import MainPage2from .pages.login_page import LoginPage3import pytest4import time5@pytest.mark.all_test6def test_user_can_log_in(setup_driver):7	link = f"https://finance.dev.fabrique.studio/"8	main_page = MainPage(setup_driver, link)9	main_page.should_be_tabel_in_form()10	main_page.should_be_btn_in_main_page()11	main_page.should_be_go_to_the_next_table()12@pytest.mark.all_test13def test_user_can_edit_and_check_payment(setup_driver):14	link = f"https://finance.dev.fabrique.studio/"15	main_page = MainPage(setup_driver, link)16	main_page.create_new_payment()17	main_page.should_be_add_payment_form_form()18	save_description = main_page.add_description_field()19	main_page.click_on_create_new_payment_in_payment_form()20	save_url = main_page.save_url_number()21	main_page.should_be_succesful_message()22	main_page.go_to_the_main_page_of_the_breadcrumbs()23	main_page.open()24	main_page.should_be_search_field()25	main_page.search_for_a_description(save_description)26	main_page.go_to_the_edit_payment()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.