Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _detach_volume method in tempest

Best Python code snippet using tempest_python

test_volumes.py

Source:test_volumes.py

1# Copyright 2013 Huawei Technologies Co.,LTD.2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15import mock16from rally.plugins.openstack.scenarios.cinder import volumes17from tests.unit import test18CINDER_VOLUMES = ("rally.plugins.openstack.scenarios.cinder.volumes"19                  ".CinderVolumes")20class fake_type(object):21    name = "fake"22class CinderServersTestCase(test.ScenarioTestCase):23    def _get_context(self):24        return {"user": {"tenant_id": "fake",25                         "endpoint": mock.MagicMock()},26                "tenant": {"id": "fake", "name": "fake",27                           "volumes": [{"id": "uuid"}],28                           "servers": [1]}}29    def test_create_and_list_volume(self):30        scenario = volumes.CinderVolumes()31        scenario._create_volume = mock.MagicMock()32        scenario._list_volumes = mock.MagicMock()33        scenario.create_and_list_volume(1, True, fakearg="f")34        scenario._create_volume.assert_called_once_with(1, fakearg="f")35        scenario._list_volumes.assert_called_once_with(True)36    def test_list_volumes(self):37        scenario = volumes.CinderVolumes()38        scenario._list_volumes = mock.MagicMock()39        scenario.list_volumes(True)40        scenario._list_volumes.assert_called_once_with(True)41    def test_create_and_delete_volume(self):42        fake_volume = mock.MagicMock()43        scenario = volumes.CinderVolumes()44        scenario._create_volume = mock.MagicMock(return_value=fake_volume)45        scenario.sleep_between = mock.MagicMock()46        scenario._delete_volume = mock.MagicMock()47        scenario.create_and_delete_volume(size=1, min_sleep=10, max_sleep=20,48                                          fakearg="f")49        scenario._create_volume.assert_called_once_with(1, fakearg="f")50        scenario.sleep_between.assert_called_once_with(10, 20)51        scenario._delete_volume.assert_called_once_with(fake_volume)52    def test_create_volume(self):53        fake_volume = mock.MagicMock()54        scenario = volumes.CinderVolumes()55        scenario._create_volume = mock.MagicMock(return_value=fake_volume)56        scenario.create_volume(1, fakearg="f")57        scenario._create_volume.assert_called_once_with(1, fakearg="f")58    def test_create_volume_and_modify_metadata(self):59        scenario = volumes.CinderVolumes(self._get_context())60        scenario._set_metadata = mock.Mock()61        scenario._delete_metadata = mock.Mock()62        scenario.modify_volume_metadata(sets=5, set_size=4,63                                        deletes=3, delete_size=2)64        scenario._set_metadata.assert_called_once_with("uuid", 5, 4)65        scenario._delete_metadata.assert_called_once_with(66            "uuid",67            scenario._set_metadata.return_value, 3, 2)68    def test_create_and_extend_volume(self):69        fake_volume = mock.MagicMock()70        scenario = volumes.CinderVolumes()71        scenario._create_volume = mock.MagicMock(return_value=fake_volume)72        scenario._extend_volume = mock.MagicMock(return_value=fake_volume)73        scenario.sleep_between = mock.MagicMock()74        scenario._delete_volume = mock.MagicMock()75        scenario.create_and_extend_volume(1, 2, 10, 20, fakearg="f")76        scenario._create_volume.assert_called_once_with(1, fakearg="f")77        self.assertTrue(scenario._extend_volume.called)78        scenario.sleep_between.assert_called_once_with(10, 20)79        scenario._delete_volume.assert_called_once_with(fake_volume)80    def test_create_from_image_and_delete_volume(self):81        fake_volume = mock.MagicMock()82        scenario = volumes.CinderVolumes()83        scenario._create_volume = mock.MagicMock(return_value=fake_volume)84        scenario._delete_volume = mock.MagicMock()85        scenario.create_and_delete_volume(1, image="fake_image")86        scenario._create_volume.assert_called_once_with(1,87                                                        imageRef="fake_image")88        scenario._delete_volume.assert_called_once_with(fake_volume)89    def test_create_volume_from_image(self):90        fake_volume = mock.MagicMock()91        scenario = volumes.CinderVolumes()92        scenario._create_volume = mock.MagicMock(return_value=fake_volume)93        scenario.create_volume(1, image="fake_image")94        scenario._create_volume.assert_called_once_with(1,95                                                        imageRef="fake_image")96    def test_create_volume_from_image_and_list(self):97        fake_volume = mock.MagicMock()98        scenario = volumes.CinderVolumes()99        scenario._create_volume = mock.MagicMock(return_value=fake_volume)100        scenario._list_volumes = mock.MagicMock()101        scenario.create_and_list_volume(1, True, "fake_image")102        scenario._create_volume.assert_called_once_with(1,103                                                        imageRef="fake_image")104        scenario._list_volumes.assert_called_once_with(True)105    def test_create_from_volume_and_delete_volume(self):106        fake_volume = mock.MagicMock()107        vol_size = 1108        scenario = volumes.CinderVolumes(self._get_context())109        scenario._create_volume = mock.MagicMock(return_value=fake_volume)110        scenario._delete_volume = mock.MagicMock()111        scenario.create_from_volume_and_delete_volume(vol_size)112        scenario._create_volume.assert_called_once_with(1, source_volid="uuid")113        scenario._delete_volume.assert_called_once_with(fake_volume)114    def test_create_and_delete_snapshot(self):115        fake_snapshot = mock.MagicMock()116        scenario = volumes.CinderVolumes(self._get_context())117        scenario._create_snapshot = mock.MagicMock(return_value=fake_snapshot)118        scenario.sleep_between = mock.MagicMock()119        scenario._delete_snapshot = mock.MagicMock()120        scenario.create_and_delete_snapshot(False, 10, 20, fakearg="f")121        scenario._create_snapshot.assert_called_once_with("uuid", force=False,122                                                          fakearg="f")123        scenario.sleep_between.assert_called_once_with(10, 20)124        scenario._delete_snapshot.assert_called_once_with(fake_snapshot)125    def test_create_and_list_snapshots(self):126        fake_snapshot = mock.MagicMock()127        scenario = volumes.CinderVolumes(self._get_context())128        scenario._create_snapshot = mock.MagicMock(return_value=fake_snapshot)129        scenario._list_snapshots = mock.MagicMock()130        scenario.create_and_list_snapshots(False, True, fakearg="f")131        scenario._create_snapshot.assert_called_once_with("uuid", force=False,132                                                          fakearg="f")133        scenario._list_snapshots.assert_called_once_with(True)134    def test_create_and_attach_volume(self):135        fake_volume = mock.MagicMock()136        fake_server = mock.MagicMock()137        scenario = volumes.CinderVolumes()138        scenario._attach_volume = mock.MagicMock()139        scenario._detach_volume = mock.MagicMock()140        scenario._boot_server = mock.MagicMock(return_value=fake_server)141        scenario._delete_server = mock.MagicMock()142        scenario._create_volume = mock.MagicMock(return_value=fake_volume)143        scenario._delete_volume = mock.MagicMock()144        scenario.create_and_attach_volume(10, "img", "0")145        scenario._attach_volume.assert_called_once_with(fake_server,146                                                        fake_volume)147        scenario._detach_volume.assert_called_once_with(fake_server,148                                                        fake_volume)149        scenario._delete_volume.assert_called_once_with(fake_volume)150        scenario._delete_server.assert_called_once_with(fake_server)151    def test_create_and_upload_volume_to_image(self):152        fake_volume = mock.Mock()153        fake_image = mock.Mock()154        scenario = volumes.CinderVolumes()155        scenario._create_volume = mock.MagicMock(return_value=fake_volume)156        scenario._upload_volume_to_image = mock.MagicMock(157            return_value=fake_image)158        scenario._delete_volume = mock.MagicMock()159        scenario._delete_image = mock.MagicMock()160        scenario.create_and_upload_volume_to_image(2,161                                                   container_format="fake",162                                                   disk_format="disk",163                                                   do_delete=False)164        scenario._create_volume.assert_called_once_with(2)165        scenario._upload_volume_to_image.assert_called_once_with(fake_volume,166                                                                 False,167                                                                 "fake",168                                                                 "disk")169        scenario._create_volume.reset_mock()170        scenario._upload_volume_to_image.reset_mock()171        scenario.create_and_upload_volume_to_image(1, do_delete=True)172        scenario._create_volume.assert_called_once_with(1)173        scenario._upload_volume_to_image.assert_called_once_with(fake_volume,174                                                                 False,175                                                                 "bare",176                                                                 "raw")177        scenario._delete_volume.assert_called_once_with(fake_volume)178        scenario._delete_image.assert_called_once_with(fake_image)179    def test_create_snapshot_and_attach_volume(self):180        fake_volume = mock.MagicMock()181        fake_snapshot = mock.MagicMock()182        fake_server = mock.MagicMock()183        scenario = volumes.CinderVolumes(self._get_context())184        scenario._attach_volume = mock.MagicMock()185        scenario._detach_volume = mock.MagicMock()186        scenario._boot_server = mock.MagicMock(return_value=fake_server)187        scenario._delete_server = mock.MagicMock()188        scenario._create_volume = mock.MagicMock(return_value=fake_volume)189        scenario._delete_volume = mock.MagicMock()190        scenario._create_snapshot = mock.MagicMock(return_value=fake_snapshot)191        scenario._delete_snapshot = mock.MagicMock()192        self.clients("nova").servers.get = mock.MagicMock(193            return_value=fake_server)194        scenario.create_snapshot_and_attach_volume()195        self.assertTrue(scenario._create_volume.called)196        scenario._create_snapshot.assert_called_once_with(fake_volume.id,197                                                          False)198        scenario._delete_snapshot.assert_called_once_with(fake_snapshot)199        scenario._attach_volume.assert_called_once_with(fake_server,200                                                        fake_volume)201        scenario._detach_volume.assert_called_once_with(fake_server,202                                                        fake_volume)203        scenario._delete_volume.assert_called_once_with(fake_volume)204    def test_create_snapshot_and_attach_volume_use_volume_type(self):205        fake_volume = mock.MagicMock()206        fake_snapshot = mock.MagicMock()207        fake_server = mock.MagicMock()208        scenario = volumes.CinderVolumes(self._get_context())209        scenario._attach_volume = mock.MagicMock()210        scenario._detach_volume = mock.MagicMock()211        scenario._boot_server = mock.MagicMock(return_value=fake_server)212        scenario._delete_server = mock.MagicMock()213        scenario._create_volume = mock.MagicMock(return_value=fake_volume)214        scenario._delete_volume = mock.MagicMock()215        scenario._create_snapshot = mock.MagicMock(return_value=fake_snapshot)216        scenario._delete_snapshot = mock.MagicMock()217        fake = fake_type()218        self.clients("cinder").volume_types.list = mock.MagicMock(219            return_value=[fake])220        self.clients("nova").servers.get = mock.MagicMock(221            return_value=fake_server)222        scenario.create_snapshot_and_attach_volume(volume_type=True)223        # Make sure create volume's second arg was the correct volume type.224        # fake or none (randomly selected)225        self.assertTrue(scenario._create_volume.called)226        vol_type = scenario._create_volume.call_args_list[0][1]["volume_type"]227        self.assertTrue(vol_type is fake.name or vol_type is None)228        scenario._create_snapshot.assert_called_once_with(fake_volume.id,229                                                          False)230        scenario._delete_snapshot.assert_called_once_with(fake_snapshot)231        scenario._attach_volume.assert_called_once_with(fake_server,232                                                        fake_volume)233        scenario._detach_volume.assert_called_once_with(fake_server,234                                                        fake_volume)235        scenario._delete_volume.assert_called_once_with(fake_volume)236    def test_create_nested_snapshots_and_attach_volume(self):237        fake_volume = mock.MagicMock()238        fake_snapshot = mock.MagicMock()239        scenario = volumes.CinderVolumes(context=self._get_context())240        scenario._attach_volume = mock.MagicMock()241        scenario._detach_volume = mock.MagicMock()242        scenario._delete_server = mock.MagicMock()243        scenario._create_volume = mock.MagicMock(return_value=fake_volume)244        scenario._delete_volume = mock.MagicMock()245        scenario._create_snapshot = mock.MagicMock(return_value=fake_snapshot)246        scenario._delete_snapshot = mock.MagicMock()247        scenario.create_nested_snapshots_and_attach_volume()248        volume_count = scenario._create_volume.call_count249        snapshots_count = scenario._create_snapshot.call_count250        attached_count = scenario._attach_volume.call_count251        self.assertEqual(scenario._delete_volume.call_count, volume_count)252        self.assertEqual(scenario._delete_snapshot.call_count, snapshots_count)253        self.assertEqual(scenario._detach_volume.call_count, attached_count)254    def test_create_nested_snapshots_calls_order(self):255        fake_volume1 = mock.MagicMock()256        fake_volume2 = mock.MagicMock()257        fake_snapshot1 = mock.MagicMock()258        fake_snapshot2 = mock.MagicMock()259        scenario = volumes.CinderVolumes(self._get_context())260        scenario._attach_volume = mock.MagicMock()261        scenario._detach_volume = mock.MagicMock()262        scenario._delete_server = mock.MagicMock()263        scenario._create_volume = mock.MagicMock(264            side_effect=[fake_volume1, fake_volume2])265        scenario._delete_volume = mock.MagicMock()266        scenario._create_snapshot = mock.MagicMock(267            side_effect=[fake_snapshot1, fake_snapshot2])268        scenario._delete_snapshot = mock.MagicMock()269        scenario.create_nested_snapshots_and_attach_volume(270            nested_level={"min": 2, "max": 2})271        vol_delete_calls = [mock.call(fake_volume2), mock.call(fake_volume1)]272        snap_delete_calls = [mock.call(fake_snapshot2),273                             mock.call(fake_snapshot1)]274        scenario._delete_volume.assert_has_calls(vol_delete_calls)275        scenario._delete_snapshot.assert_has_calls(snap_delete_calls)276    @mock.patch("rally.plugins.openstack.scenarios.cinder.volumes.random")277    def test_create_nested_snapshots_check_resources_size(self, mock_random):278        mock_random.randint.return_value = 3279        fake_volume = mock.MagicMock()280        fake_snapshot = mock.MagicMock()281        fake_server = mock.MagicMock()282        scenario = volumes.CinderVolumes(self._get_context())283        scenario.get_random_server = mock.MagicMock(return_value=fake_server)284        scenario._attach_volume = mock.MagicMock()285        scenario._detach_volume = mock.MagicMock()286        scenario._delete_server = mock.MagicMock()287        scenario._create_volume = mock.MagicMock(return_value=fake_volume)288        scenario._delete_volume = mock.MagicMock()289        scenario._create_snapshot = mock.MagicMock(return_value=fake_snapshot)290        scenario._delete_snapshot = mock.MagicMock()291        scenario.create_nested_snapshots_and_attach_volume()292        # NOTE: Two calls for random size and nested level293        random_call_count = mock_random.randint.call_count294        self.assertEqual(2, random_call_count)295        calls = scenario._create_volume.mock_calls296        expected_calls = [mock.call(3),297                          mock.call(3, snapshot_id=fake_snapshot.id),298                          mock.call(3, snapshot_id=fake_snapshot.id)]299        self.assertEqual(expected_calls, calls)300    def test_create_volume_backup(self):301        fake_volume = mock.MagicMock()302        fake_backup = mock.MagicMock()303        scenario = self._get_scenario(fake_volume, fake_backup)304        volume_kwargs = {"some_var": "zaq"}305        scenario.create_volume_backup(306            1, do_delete=True, create_volume_kwargs=volume_kwargs)307        scenario._create_volume.assert_called_once_with(1, **volume_kwargs)308        scenario._create_backup.assert_called_once_with(fake_volume.id)309        scenario._delete_volume.assert_called_once_with(fake_volume)310        scenario._delete_backup.assert_called_once_with(fake_backup)311    def test_create_volume_backup_no_delete(self):312        fake_volume = mock.MagicMock()313        fake_backup = mock.MagicMock()314        scenario = self._get_scenario(fake_volume, fake_backup)315        volume_kwargs = {"some_var": "zaq"}316        scenario.create_volume_backup(317            1, do_delete=False, create_volume_kwargs=volume_kwargs)318        scenario._create_volume.assert_called_once_with(1, **volume_kwargs)319        scenario._create_backup.assert_called_once_with(fake_volume.id)320        self.assertFalse(scenario._delete_volume.called)321        self.assertFalse(scenario._delete_backup.called)322    def _get_scenario(self, fake_volume, fake_backup, fake_restore=None):323        scenario = volumes.CinderVolumes()324        scenario._create_volume = mock.MagicMock(return_value=fake_volume)325        scenario._create_backup = mock.MagicMock(return_value=fake_backup)326        scenario._restore_backup = mock.MagicMock(return_value=fake_restore)327        scenario._list_backups = mock.MagicMock()328        scenario._delete_volume = mock.MagicMock()329        scenario._delete_backup = mock.MagicMock()330        return scenario331    def test_create_and_restore_volume_backup(self):332        fake_volume = mock.MagicMock()333        fake_backup = mock.MagicMock()334        fake_restore = mock.MagicMock()335        scenario = self._get_scenario(fake_volume, fake_backup, fake_restore)336        volume_kwargs = {"some_var": "zaq"}337        scenario.create_and_restore_volume_backup(338            1, do_delete=True, create_volume_kwargs=volume_kwargs)339        scenario._create_volume.assert_called_once_with(1, **volume_kwargs)340        scenario._create_backup.assert_called_once_with(fake_volume.id)341        scenario._restore_backup.assert_called_once_with(fake_backup.id)342        scenario._delete_volume.assert_called_once_with(fake_volume)343        scenario._delete_backup.assert_called_once_with(fake_backup)344    def test_create_and_restore_volume_backup_no_delete(self):345        fake_volume = mock.MagicMock()346        fake_backup = mock.MagicMock()347        fake_restore = mock.MagicMock()348        scenario = self._get_scenario(fake_volume, fake_backup, fake_restore)349        volume_kwargs = {"some_var": "zaq"}350        scenario.create_and_restore_volume_backup(351            1, do_delete=False, create_volume_kwargs=volume_kwargs)352        scenario._create_volume.assert_called_once_with(1, **volume_kwargs)353        scenario._create_backup.assert_called_once_with(fake_volume.id)354        scenario._restore_backup.assert_called_once_with(fake_backup.id)355        self.assertFalse(scenario._delete_volume.called)356        self.assertFalse(scenario._delete_backup.called)357    def test_create_and_list_volume_backups(self):358        fake_volume = mock.MagicMock()359        fake_backup = mock.MagicMock()360        scenario = self._get_scenario(fake_volume, fake_backup)361        volume_kwargs = {"some_var": "zaq"}362        scenario.create_and_list_volume_backups(363            1, detailed=True, do_delete=True,364            create_volume_kwargs=volume_kwargs)365        scenario._create_volume.assert_called_once_with(1, **volume_kwargs)366        scenario._create_backup.assert_called_once_with(fake_volume.id)367        scenario._list_backups.assert_called_once_with(True)368        scenario._delete_volume.assert_called_once_with(fake_volume)369        scenario._delete_backup.assert_called_once_with(fake_backup)370    def test_create_and_list_volume_backups_no_delete(self):371        fake_volume = mock.MagicMock()372        fake_backup = mock.MagicMock()373        scenario = self._get_scenario(fake_volume, fake_backup)374        volume_kwargs = {"some_var": "zaq"}375        scenario.create_and_list_volume_backups(376            1, detailed=True, do_delete=False,377            create_volume_kwargs=volume_kwargs)378        scenario._create_volume.assert_called_once_with(1, **volume_kwargs)379        scenario._create_backup.assert_called_once_with(fake_volume.id)380        scenario._list_backups.assert_called_once_with(True)381        self.assertFalse(scenario._delete_volume.called)...

test_volume.py

Source:test_volume.py

...83                          volume_id)84        db.volume_destroy(context.get_admin_context(), volume_id)85        for volume_id in vols:86            self.volume.delete_volume(self.context, volume_id)87    def test_run_attach_detach_volume(self):88        """Make sure volume can be attached and detached from instance."""89        inst = {}90        inst['image_id'] = 191        inst['reservation_id'] = 'r-fakeres'92        inst['launch_time'] = '10'93        inst['user_id'] = 'fake'94        inst['project_id'] = 'fake'95        inst['instance_type_id'] = '2'  # m1.tiny96        inst['mac_address'] = utils.generate_mac()97        inst['ami_launch_index'] = 098        instance_id = db.instance_create(self.context, inst)['id']99        mountpoint = "/dev/sdf"100        volume_id = self._create_volume()101        self.volume.create_volume(self.context, volume_id)102        if FLAGS.fake_tests:103            db.volume_attached(self.context, volume_id, instance_id,104                               mountpoint)105        else:106            self.compute.attach_volume(self.context,107                                       instance_id,108                                       volume_id,109                                       mountpoint)110        vol = db.volume_get(context.get_admin_context(), volume_id)111        self.assertEqual(vol['status'], "in-use")112        self.assertEqual(vol['attach_status'], "attached")113        self.assertEqual(vol['mountpoint'], mountpoint)114        instance_ref = db.volume_get_instance(self.context, volume_id)115        self.assertEqual(instance_ref['id'], instance_id)116        self.assertRaises(exception.Error,117                          self.volume.delete_volume,118                          self.context,119                          volume_id)120        if FLAGS.fake_tests:121            db.volume_detached(self.context, volume_id)122        else:123            self.compute.detach_volume(self.context,124                                       instance_id,125                                       volume_id)126        vol = db.volume_get(self.context, volume_id)127        self.assertEqual(vol['status'], "available")128        self.volume.delete_volume(self.context, volume_id)129        self.assertRaises(exception.VolumeNotFound,130                          db.volume_get,131                          self.context,132                          volume_id)133        db.instance_destroy(self.context, instance_id)134    def test_concurrent_volumes_get_different_targets(self):135        """Ensure multiple concurrent volumes get different targets."""136        volume_ids = []137        targets = []138        def _check(volume_id):139            """Make sure targets aren't duplicated."""140            volume_ids.append(volume_id)141            admin_context = context.get_admin_context()142            iscsi_target = db.volume_get_iscsi_target_num(admin_context,143                                                          volume_id)144            self.assert_(iscsi_target not in targets)145            targets.append(iscsi_target)146            LOG.debug(_("Target %s allocated"), iscsi_target)147        total_slots = FLAGS.iscsi_num_targets148        for _index in xrange(total_slots):149            volume_id = self._create_volume()150            d = self.volume.create_volume(self.context, volume_id)151            _check(d)152        for volume_id in volume_ids:153            self.volume.delete_volume(self.context, volume_id)154    def test_multi_node(self):155        # TODO(termie): Figure out how to test with two nodes,156        # each of them having a different FLAG for storage_node157        # This will allow us to test cross-node interactions158        pass159class DriverTestCase(test.TestCase):160    """Base Test class for Drivers."""161    driver_name = "nova.volume.driver.FakeAOEDriver"162    def setUp(self):163        super(DriverTestCase, self).setUp()164        self.flags(volume_driver=self.driver_name,165                   logging_default_format_string="%(message)s")166        self.volume = utils.import_object(FLAGS.volume_manager)167        self.context = context.get_admin_context()168        self.output = ""169        def _fake_execute(_command, *_args, **_kwargs):170            """Fake _execute."""171            return self.output, None172        self.volume.driver._execute = _fake_execute173        self.volume.driver._sync_execute = _fake_execute174        log = logging.getLogger()175        self.stream = cStringIO.StringIO()176        log.addHandler(logging.StreamHandler(self.stream))177        inst = {}178        self.instance_id = db.instance_create(self.context, inst)['id']179    def tearDown(self):180        super(DriverTestCase, self).tearDown()181    def _attach_volume(self):182        """Attach volumes to an instance. This function also sets183           a fake log message."""184        return []185    def _detach_volume(self, volume_id_list):186        """Detach volumes from an instance."""187        for volume_id in volume_id_list:188            db.volume_detached(self.context, volume_id)189            self.volume.delete_volume(self.context, volume_id)190class AOETestCase(DriverTestCase):191    """Test Case for AOEDriver"""192    driver_name = "nova.volume.driver.AOEDriver"193    def setUp(self):194        super(AOETestCase, self).setUp()195    def tearDown(self):196        super(AOETestCase, self).tearDown()197    def _attach_volume(self):198        """Attach volumes to an instance. This function also sets199           a fake log message."""200        volume_id_list = []201        for index in xrange(3):202            vol = {}203            vol['size'] = 0204            volume_id = db.volume_create(self.context,205                                         vol)['id']206            self.volume.create_volume(self.context, volume_id)207            # each volume has a different mountpoint208            mountpoint = "/dev/sd" + chr((ord('b') + index))209            db.volume_attached(self.context, volume_id, self.instance_id,210                               mountpoint)211            (shelf_id, blade_id) = db.volume_get_shelf_and_blade(self.context,212                                                                 volume_id)213            self.output += "%s %s eth0 /dev/nova-volumes/vol-foo auto run\n" \214                      % (shelf_id, blade_id)215            volume_id_list.append(volume_id)216        return volume_id_list217    def test_check_for_export_with_no_volume(self):218        """No log message when no volume is attached to an instance."""219        self.stream.truncate(0)220        self.volume.check_for_export(self.context, self.instance_id)221        self.assertEqual(self.stream.getvalue(), '')222    def test_check_for_export_with_all_vblade_processes(self):223        """No log message when all the vblade processes are running."""224        volume_id_list = self._attach_volume()225        self.stream.truncate(0)226        self.volume.check_for_export(self.context, self.instance_id)227        self.assertEqual(self.stream.getvalue(), '')228        self._detach_volume(volume_id_list)229    def test_check_for_export_with_vblade_process_missing(self):230        """Output a warning message when some vblade processes aren't231           running."""232        volume_id_list = self._attach_volume()233        # the first vblade process isn't running234        self.output = self.output.replace("run", "down", 1)235        (shelf_id, blade_id) = db.volume_get_shelf_and_blade(self.context,236                                                             volume_id_list[0])237        msg_is_match = False238        self.stream.truncate(0)239        try:240            self.volume.check_for_export(self.context, self.instance_id)241        except exception.ProcessExecutionError, e:242            volume_id = volume_id_list[0]243            msg = _("Cannot confirm exported volume id:%(volume_id)s. "244                    "vblade process for e%(shelf_id)s.%(blade_id)s "245                    "isn't running.") % locals()246            msg_is_match = (0 <= e.message.find(msg))247        self.assertTrue(msg_is_match)248        self._detach_volume(volume_id_list)249class ISCSITestCase(DriverTestCase):250    """Test Case for ISCSIDriver"""251    driver_name = "nova.volume.driver.ISCSIDriver"252    def setUp(self):253        super(ISCSITestCase, self).setUp()254    def tearDown(self):255        super(ISCSITestCase, self).tearDown()256    def _attach_volume(self):257        """Attach volumes to an instance. This function also sets258           a fake log message."""259        volume_id_list = []260        for index in xrange(3):261            vol = {}262            vol['size'] = 0263            vol_ref = db.volume_create(self.context, vol)264            self.volume.create_volume(self.context, vol_ref['id'])265            vol_ref = db.volume_get(self.context, vol_ref['id'])266            # each volume has a different mountpoint267            mountpoint = "/dev/sd" + chr((ord('b') + index))268            db.volume_attached(self.context, vol_ref['id'], self.instance_id,269                               mountpoint)270            volume_id_list.append(vol_ref['id'])271        return volume_id_list272    def test_check_for_export_with_no_volume(self):273        """No log message when no volume is attached to an instance."""274        self.stream.truncate(0)275        self.volume.check_for_export(self.context, self.instance_id)276        self.assertEqual(self.stream.getvalue(), '')277    def test_check_for_export_with_all_volume_exported(self):278        """No log message when all the vblade processes are running."""279        volume_id_list = self._attach_volume()280        self.mox.StubOutWithMock(self.volume.driver, '_execute')281        for i in volume_id_list:282            tid = db.volume_get_iscsi_target_num(self.context, i)283            self.volume.driver._execute("sudo", "ietadm", "--op", "show",284                                        "--tid=%(tid)d" % locals())285        self.stream.truncate(0)286        self.mox.ReplayAll()287        self.volume.check_for_export(self.context, self.instance_id)288        self.assertEqual(self.stream.getvalue(), '')289        self.mox.UnsetStubs()290        self._detach_volume(volume_id_list)291    def test_check_for_export_with_some_volume_missing(self):292        """Output a warning message when some volumes are not recognied293           by ietd."""294        volume_id_list = self._attach_volume()295        # the first vblade process isn't running296        tid = db.volume_get_iscsi_target_num(self.context, volume_id_list[0])297        self.mox.StubOutWithMock(self.volume.driver, '_execute')298        self.volume.driver._execute("sudo", "ietadm", "--op", "show",299                                    "--tid=%(tid)d" % locals()).AndRaise(300                                            exception.ProcessExecutionError())301        self.mox.ReplayAll()302        self.assertRaises(exception.ProcessExecutionError,303                          self.volume.check_for_export,304                          self.context,305                          self.instance_id)306        msg = _("Cannot confirm exported volume id:%s.") % volume_id_list[0]307        self.assertTrue(0 <= self.stream.getvalue().find(msg))308        self.mox.UnsetStubs()...

test_ebs.py

Source:test_ebs.py

1'''2Created on Sep 4, 20123 4@author: marat5'''6 7import mock8from nose.tools import raises9 10from scalarizr.storage2 import StorageError11from scalarizr.storage2.volumes import base12from scalarizr.storage2.volumes import ebs13from scalarizr.linux import coreutils14 15 16def test_name2device():17    pass18 19 20@mock.patch('os.path.exists', return_value=True)21def test_name2device_xen(*args):22    device = ebs.name2device('/dev/sda1')23    assert device == '/dev/xvda1'24 25 26@mock.patch('os.path.exists', return_value=True)27@mock.patch.object(ebs, 'storage2')28def test_name2device_rhel_bug(s, exists):29    s.RHEL_DEVICE_ORDERING_BUG = True30    device = ebs.name2device('/dev/sda1')31    assert device == '/dev/xvde1'32 33 34def test_name2device_device_passed():35    pass36 37def test_device2name():38    pass39 40def test_device2name_xen():41    pass42 43def test_device2name_rhel_bug():44    pass45 46def test_device2name_name_passed():47    pass48 49 50class TestFreeDeviceLetterMgr(object):51    def setup(self):52        self.mgr = ebs.FreeDeviceLetterMgr()53 54    @mock.patch('glob.glob')55    @mock.patch.object(ebs, '__node__', new={'ec2': {56                                    't1micro_detached_ebs': None}})57    def test_acquire(self, glob):58        glob_returns = [['/dev/sdf1'], []]59        def globfn(*args, **kwds):60            return glob_returns.pop(0)61        glob.side_effect = globfn62 63        letter = self.mgr.__enter__().get()64 65        glob.assert_called_with('/dev/sd%s*' % letter)66        assert glob.call_count == 267 68 69    def test_acquire_concurrent(self):70        pass71 72 73    @mock.patch('glob.glob', return_value=[])74    @mock.patch.object(ebs, '__node__', new={'ec2': {75                                    't1micro_detached_ebs': ['/dev/sdg', '/dev/sdf']}})76    def test_acquire_t1micro(self, glob):77        letter = self.mgr.__enter__().get()78        assert letter not in ('g', 'f')79 80 81Ebs = ebs.EbsVolume82@mock.patch.object(ebs, '__node__', new={'ec2': {83                                'instance_id': 'i-12345678',84                                'instance_type': 'm1.small',85                                'avail_zone': 'us-east-1a'}})86@mock.patch.object(ebs, 'name2device',87                                side_effect=lambda name: name.replace('/sd', '/xvd'))88@mock.patch.object(Ebs, '_free_device_letter_mgr',89                                **{'get.return_value' : 'b'})90@mock.patch.object(Ebs, '_connect_ec2')91class TestEbsVolume(object):92 93    @mock.patch.object(Ebs, '_attach_volume')94    @mock.patch.object(Ebs, '_create_volume')95    def test_ensure_new(self, _create_volume, *args):96        ebs = mock.Mock(97                id='vol-12345678',98                size=1,99                zone='us-east-1a',100                **{'volume_state.return_value': 'available',101                        'attach_data.device': '/dev/sdb'}102        )103        _create_volume.return_value = ebs104 105        self.vol = Ebs(type='ebs')106        self.vol.ensure()107 108        assert self.vol.id == ebs.id109        assert self.vol.size == ebs.size110        assert self.vol.avail_zone == 'us-east-1a'111        assert self.vol.name == '/dev/sdb'112        assert self.vol.device == '/dev/xvdb'113        assert self.vol.config()114 115 116    @mock.patch.object(Ebs, '_attach_volume')117    def test_ensure_existed(self, av, _connect_ec2, *args):118        conn = _connect_ec2.return_value119        ebs = mock.Mock(120                id='vol-12345678',121                size=1,122                zone='us-east-1a',123                **{'volume_state.return_value': 'available'}124        )125        conn.get_all_volumes.return_value = [ebs]126 127        vol = Ebs(type='ebs', id='vol-12345678')128        vol.ensure()129 130        assert vol.id == ebs.id131        assert vol.size == ebs.size132        assert vol.avail_zone == 'us-east-1a'133        assert vol.name == '/dev/sdb'134        assert vol.device == '/dev/xvdb'135        conn.get_all_volumes.assert_called_once_with(['vol-12345678'])136        assert vol.config()137 138 139    @mock.patch.object(Ebs, '_attach_volume')140    @mock.patch.object(Ebs, '_create_volume')141    @mock.patch.object(Ebs, '_create_snapshot')142    def test_ensure_existed_in_different_zone(self, _create_snapshot,143                            _create_volume, av, _connect_ec2, *args):144        conn = _connect_ec2.return_value145        ebs = mock.Mock(146                id='vol-12345678',147                size=1,148                zone='us-east-1b',149                **{'volume_state.return_value': 'available'}150        )151        ebs2 = mock.Mock(152                id='vol-87654321',153                size=1,154                zone='us-east-1a',155                **{'volume_state.return_value': 'available'}156        )157        conn.get_all_volumes.return_value = [ebs]158        _create_volume.return_value = ebs2159        _create_snapshot.return_value = mock.Mock(id='snap-12345678')160 161        vol = Ebs(type='ebs', id='vol-12345678')162        vol.ensure()163 164        assert vol.id == ebs2.id165 166 167    @mock.patch.object(Ebs, '_attach_volume')168    @mock.patch.object(Ebs, '_detach_volume')169    def test_ensure_existed_attached_to_other_instance(self, av, dv,170                                    _connect_ec2, *args):171        conn = _connect_ec2.return_value172        ebs = mock.Mock(173                id='vol-12345678',174                size=1,175                zone='us-east-1a',176                **{'volume_state.return_value': 'available',177                        'attachment_state.return_value': 'attached'}178        )179        conn.get_all_volumes.return_value = [ebs]180 181        vol = Ebs(type='ebs', id='vol-12345678')182        vol.ensure()183 184        assert vol.id == ebs.id185        vol._detach_volume.assert_called_once_with(ebs)186        vol._attach_volume.assert_called_once_with(ebs, '/dev/sdb')187 188 189    @mock.patch.object(Ebs, '_attach_volume')190    @mock.patch.object(Ebs, '_detach_volume')191    @mock.patch.object(Ebs, '_wait_attachment_state_change')192    def test_ensure_existed_attaching_to_other_instance(self,193                                    av, dv, wasc, _connect_ec2, *args):194        ebs = mock.Mock(195                id='vol-12345678',196                size=1,197                zone='us-east-1a',198                **{'volume_state.return_value': 'available',199                        'attachment_state.return_value': 'attaching'}200        )201        _connect_ec2.return_value.get_all_volumes.return_value = [ebs]202 203        vol = Ebs(type='ebs', id='vol-12345678')204        vol.ensure()205 206        assert vol.id == ebs.id207        assert vol._detach_volume.call_count == 0208        vol._attach_volume.assert_called_once_with(ebs, '/dev/sdb')209 210 211    @mock.patch.object(Ebs, '_create_volume')212    @mock.patch.object(Ebs, '_attach_volume')213    def test_ensure_restore(self, vc, av, *args):214        snap_config = {215                'id':'snap-12345678',216                'tags':{'ta':'gs'}217        }218        size = 1219        vol = Ebs(size=size, snap=snap_config)220        vol.ensure()221 222        vol._create_volume.assert_called_once_with(223                        zone=mock.ANY,224                        size=size,225                        snapshot=snap_config['id'],226                        tags=snap_config['tags'],227                        volume_type=mock.ANY,228                        iops=mock.ANY)229 230 231    @mock.patch.object(Ebs, '_create_snapshot')232    def test_snapshot(self, cs, _connect_ec2, *args):233        description='test'234        tags={'ta':'gs'}235        vol = Ebs(type='ebs', id='vol-12345678')236 237        vol.snapshot(description, tags=tags, nowait=True)238        vol._create_snapshot.assert_called_once_with(239                                vol.id, description, tags, True)240 241 242    @mock.patch.object(coreutils, 'sync')243    def test_snapshot_tags(self, sync, _connect_ec2, *args):244        snapshot = mock.Mock(id='snap-12345678')245        conn = _connect_ec2.return_value246        conn.configure_mock(**{'create_snapshot.return_value': snapshot,247                                                        'create_tags': mock.Mock()})248 249        tags = {'ta':'gs'}250        vol = Ebs(type='ebs', id='vol-12345678')251        snap = vol.snapshot('test', tags=tags)252 253        conn.create_tags.assert_called_once_with([snapshot.id], tags)254 255 256    @mock.patch.object(Ebs, '_wait_snapshot')257    def test_snapshot_wait(self, ws, _connect_ec2, *args):258        snapshot = mock.Mock(id='snap-12345678')259        _connect_ec2.return_value.configure_mock(**{260                        'create_snapshot.return_value': snapshot,261                        'create_tags': mock.Mock()})262 263        vol = Ebs(type='ebs', id='vol-12345678', tags={'ta': 'gs'})264        snap = vol.snapshot('test', tags=vol.tags, nowait=False)265 266        vol._wait_snapshot.assert_called_once_with(snapshot)267 268 269    @raises(AssertionError)270    def test_snapshot_no_connection(self, _connect_ec2, *args):271        _connect_ec2.return_value = None272        self.vol = Ebs(type='ebs', id='vol-12345678')273        snap = self.vol.snapshot('test')274 275 276    @mock.patch.object(Ebs, 'umount')277    @mock.patch.object(Ebs, '_detach_volume')278    def test_detach(self, *args):279        vol = Ebs(type='ebs', device='/dev/xvdp', id='vol-12345678')280        vol.detach()281        vol._detach_volume.assert_called_with(vol.id, False)282 283 284    @mock.patch.object(Ebs, '_detach_volume')285    @mock.patch.object(Ebs, '_instance_type', return_value='t1.micro')286    def test_detach_t1micro(self, *args):287        ebs.__node__['ec2']['t1micro_detached_ebs'] = None288        vol = Ebs(name='/dev/sdf', type='ebs', id='vol-12345678')289        vol._detach(True)290        assert ebs.__node__['ec2']['t1micro_detached_ebs'] == [vol.name,]291 292 293    @raises(AssertionError)294    def test_detach_no_connection(self, _connect_ec2, *args):295        _connect_ec2.return_value = None296        vol = Ebs(type='ebs', id='vol-12345678', device='/dev/xvdp')297        vol._detach(False)298 299 300    def test_destroy(self, _connect_ec2, *args):301        vol = Ebs(type='ebs', id='vol-12345678')302        vol.destroy(True)303        conn = _connect_ec2.return_value304        conn.delete_volume.assert_called_once_with(vol.id)305 306 307    @raises(AssertionError)308    def test_destroy_no_connection(self, _connect_ec2, *args):309        _connect_ec2.return_value = None310        vol = Ebs(type='ebs', id='vol-12345678')311        vol.destroy(True)312 313 314EbsSnapshot = ebs.EbsSnapshot315@mock.patch.object(EbsSnapshot, '_connect_ec2')316class TestEbsSnapshot(object):317 318    @mock.patch.object(EbsSnapshot, '_ebs_snapshot')319    def test_status(self, _ebs_snapshot, _connect_ec2, *args, **kwargs):320        snap = EbsSnapshot(id='vol-123456ab')321        snapshot = mock.Mock(id='vol-123456ab')322        snapshot.status = 'pending'323        _ebs_snapshot.return_value = snapshot324        assert snap.status() == base.Snapshot.IN_PROGRESS325        _ebs_snapshot.assert_called_once_with('vol-123456ab')326        snapshot.update.assert_called_with()327        snapshot.status = 'completed'328        assert snap.status() == base.Snapshot.COMPLETED329        snapshot.status = 'error'330        assert snap.status() == base.Snapshot.FAILED331 332 333    @raises(AssertionError)334    def test_status_no_connection(self, _connect_ec2, *args, **kwargs):335        _connect_ec2.return_value = None336        snap = EbsSnapshot(id='vol-123456ab')337        snap.status()338 339 340    def test_destroy(self, _connect_ec2, *args, **kwargs):341        snap = EbsSnapshot(id='vol-123456ab')342        snap.destroy()343        conn = _connect_ec2.return_value344        conn.delete_snapshot.assert_called_once_with(snap.id)...

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.