FM / cloud

/opt/imh-python/lib/python3.9/site-packages/openstack/tests/functional/cloud/

/opt/imh-python/lib/python3.9/site-packages/openstack/tests/functional/cloud

Name	Size	Mode	Actions
hooks/	-	0755	rm
__pycache__/	-	0755	rm
test_aggregate.py	2091	0644	edit dl rm
test_clustering.py	49762	0644	edit dl rm
test_cluster_templates.py	4186	0644	edit dl rm
test_coe_clusters.py	1013	0644	edit dl rm
test_compute.py	22058	0644	edit dl rm
test_devstack.py	1892	0644	edit dl rm
test_domain.py	5101	0644	edit dl rm
test_endpoints.py	7841	0644	edit dl rm
test_flavor.py	6683	0644	edit dl rm
test_floating_ip.py	11651	0644	edit dl rm
test_floating_ip_pool.py	1792	0644	edit dl rm
test_groups.py	4017	0644	edit dl rm
test_identity.py	10600	0644	edit dl rm
test_image.py	6827	0644	edit dl rm
test_inventory.py	3783	0644	edit dl rm
test_keypairs.py	2424	0644	edit dl rm
test_limits.py	1840	0644	edit dl rm
test_magnum_services.py	1451	0644	edit dl rm
test_network.py	5327	0644	edit dl rm
test_object.py	7450	0644	edit dl rm
test_port.py	5865	0644	edit dl rm
test_project.py	4796	0644	edit dl rm
test_project_cleanup.py	3991	0644	edit dl rm
test_qos_bandwidth_limit_rule.py	4110	0644	edit dl rm
test_qos_dscp_marking_rule.py	2775	0644	edit dl rm
test_qos_minimum_bandwidth_rule.py	2808	0644	edit dl rm
test_qos_policy.py	3997	0644	edit dl rm
test_quotas.py	3242	0644	edit dl rm
test_range_search.py	6120	0644	edit dl rm
test_recordset.py	6629	0644	edit dl rm
test_router.py	13924	0644	edit dl rm
test_security_groups.py	2536	0644	edit dl rm
test_server_group.py	1449	0644	edit dl rm
test_services.py	5332	0644	edit dl rm
test_stack.py	5737	0644	edit dl rm
test_users.py	6634	0644	edit dl rm
test_volume.py	6805	0644	edit dl rm
test_volume_backup.py	5081	0644	edit dl rm
test_volume_type.py	4567	0644	edit dl rm
test_zone.py	3079	0644	edit dl rm
util.py	1358	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/openstack/tests/functional/cloud/test_volume.py (6805B)

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

"""
test_volume
----------------------------------

Functional tests for `shade` block storage methods.
"""

from fixtures import TimeoutException
from testtools import content

from openstack.cloud import exc
from openstack.tests.functional import base
from openstack import utils

class TestVolume(base.BaseFunctionalTest):

# Creating and deleting volumes is slow
    TIMEOUT_SCALING_FACTOR = 1.5

def setUp(self):
        super(TestVolume, self).setUp()
        self.skipTest('Volume functional tests temporarily disabled')
        if not self.user_cloud.has_service('volume'):
            self.skipTest('volume service not supported by cloud')

def test_volumes(self):
        '''Test volume and snapshot functionality'''
        volume_name = self.getUniqueString()
        snapshot_name = self.getUniqueString()
        self.addDetail('volume', content.text_content(volume_name))
        self.addCleanup(self.cleanup, volume_name, snapshot_name=snapshot_name)
        volume = self.user_cloud.create_volume(
            display_name=volume_name, size=1)
        snapshot = self.user_cloud.create_volume_snapshot(
            volume['id'],
            display_name=snapshot_name
        )

ret_volume = self.user_cloud.get_volume_by_id(volume['id'])
        self.assertEqual(volume['id'], ret_volume['id'])

volume_ids = [v['id'] for v in self.user_cloud.list_volumes()]
        self.assertIn(volume['id'], volume_ids)

snapshot_list = self.user_cloud.list_volume_snapshots()
        snapshot_ids = [s['id'] for s in snapshot_list]
        self.assertIn(snapshot['id'], snapshot_ids)

ret_snapshot = self.user_cloud.get_volume_snapshot_by_id(
            snapshot['id'])
        self.assertEqual(snapshot['id'], ret_snapshot['id'])

self.user_cloud.delete_volume_snapshot(snapshot_name, wait=True)
        self.user_cloud.delete_volume(volume_name, wait=True)

def test_volume_to_image(self):
        '''Test volume export to image functionality'''
        volume_name = self.getUniqueString()
        image_name = self.getUniqueString()
        self.addDetail('volume', content.text_content(volume_name))
        self.addCleanup(self.cleanup, volume_name, image_name=image_name)
        volume = self.user_cloud.create_volume(
            display_name=volume_name, size=1)
        image = self.user_cloud.create_image(
            image_name, volume=volume, wait=True)

volume_ids = [v['id'] for v in self.user_cloud.list_volumes()]
        self.assertIn(volume['id'], volume_ids)

image_list = self.user_cloud.list_images()
        image_ids = [s['id'] for s in image_list]
        self.assertIn(image['id'], image_ids)

self.user_cloud.delete_image(image_name, wait=True)
        self.user_cloud.delete_volume(volume_name, wait=True)

def cleanup(self, volume, snapshot_name=None, image_name=None):
        # Need to delete snapshots before volumes
        if snapshot_name:
            snapshot = self.user_cloud.get_volume_snapshot(snapshot_name)
            if snapshot:
                self.user_cloud.delete_volume_snapshot(
                    snapshot_name, wait=True)
        if image_name:
            image = self.user_cloud.get_image(image_name)
            if image:
                self.user_cloud.delete_image(image_name, wait=True)
        if not isinstance(volume, list):
            self.user_cloud.delete_volume(volume, wait=True)
        else:
            # We have more than one volume to clean up - submit all of the
            # deletes without wait, then poll until none of them are found
            # in the volume list anymore
            for v in volume:
                self.user_cloud.delete_volume(v, wait=False)
            try:
                for count in utils.iterate_timeout(
                        180, "Timeout waiting for volume cleanup"):
                    found = False
                    for existing in self.user_cloud.list_volumes():
                        for v in volume:
                            if v['id'] == existing['id']:
                                found = True
                                break
                        if found:
                            break
                    if not found:
                        break
            except (exc.OpenStackCloudTimeout, TimeoutException):
                # NOTE(slaweq): ups, some volumes are still not removed
                # so we should try to force delete it once again and move
                # forward
                for existing in self.user_cloud.list_volumes():
                    for v in volume:
                        if v['id'] == existing['id']:
                            self.operator_cloud.delete_volume(
                                v, wait=False, force=True)

def test_list_volumes_pagination(self):
        '''Test pagination for list volumes functionality'''

volumes = []
        # the number of created volumes needs to be higher than
        # CONF.osapi_max_limit but not higher than volume quotas for
        # the test user in the tenant(default quotas is set to 10)
        num_volumes = 8
        for i in range(num_volumes):
            name = self.getUniqueString()
            v = self.user_cloud.create_volume(display_name=name, size=1)
            volumes.append(v)
        self.addCleanup(self.cleanup, volumes)
        result = []
        for i in self.user_cloud.list_volumes():
            if i['name'] and i['name'].startswith(self.id()):
                result.append(i['id'])
        self.assertEqual(
            sorted([i['id'] for i in volumes]),
            sorted(result))

def test_update_volume(self):
        name, desc = self.getUniqueString('name'), self.getUniqueString('desc')
        self.addCleanup(self.cleanup, name)
        volume = self.user_cloud.create_volume(1, name=name, description=desc)
        self.assertEqual(volume.name, name)
        self.assertEqual(volume.description, desc)
        new_name = self.getUniqueString('name')
        volume = self.user_cloud.update_volume(volume.id, name=new_name)
        self.assertNotEqual(volume.name, name)
        self.assertEqual(volume.name, new_name)
        self.assertEqual(volume.description, desc)

cmd: