# Copyright 2012 United States Government as represented by the # Administrator of the National Aeronautics and Space Administration. # All Rights Reserved. # # Copyright 2012 Nebula, Inc. # # 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. import django from django.core.urlresolvers import reverse from django.forms import widgets from django import http from django.template.defaultfilters import slugify # noqa from django.test.utils import override_settings from django.utils.http import urlunquote from mox3.mox import IsA # noqa import six from six import moves from openstack_dashboard import api from openstack_dashboard.api import cinder from openstack_dashboard.test import helpers as test from openstack_dashboard.usage import quotas VOLUME_INDEX_URL = reverse('horizon:project:volumes:index') VOLUME_VOLUMES_TAB_URL = urlunquote(reverse( 'horizon:project:volumes:volumes_tab')) SEARCH_OPTS = dict(status=api.cinder.VOLUME_STATE_AVAILABLE) class VolumeViewTests(test.ResetImageAPIVersionMixin, test.TestCase): @test.create_stubs({cinder: ('volume_create', 'volume_snapshot_list', 'volume_type_list', 'volume_type_default', 'volume_list', 'volume_list_paged', 'availability_zone_list', 'extension_supported'), api.glance: ('image_list_detailed',), quotas: ('tenant_limit_usages',)}) def test_create_volume(self): volume = self.cinder_volumes.first() volume_type = self.cinder_volume_types.first() az = self.cinder_availability_zones.first().zoneName usage_limit = {'maxTotalVolumeGigabytes': 250, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'type': volume_type.name, 'size': 50, 'snapshot_source': '', 'availability_zone': az} cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) cinder.volume_create(IsA(http.HttpRequest), formData['size'], formData['name'], formData['description'], formData['type'], metadata={}, snapshot_id=None, image_id=None, availability_zone=formData['availability_zone'], source_volid=None)\ .AndReturn(volume) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') res = self.client.post(url, formData) self.assertNoFormErrors(res) redirect_url = VOLUME_VOLUMES_TAB_URL self.assertRedirectsNoFollow(res, redirect_url) @test.create_stubs({cinder: ('volume_create', 'volume_snapshot_list', 'volume_type_list', 'volume_list', 'volume_type_default', 'availability_zone_list', 'extension_supported'), api.glance: ('image_list_detailed',), quotas: ('tenant_limit_usages',)}) def test_create_volume_without_name(self): volume = self.cinder_volumes.first() volume_type = self.cinder_volume_types.first() az = self.cinder_availability_zones.first().zoneName usage_limit = {'maxTotalVolumeGigabytes': 250, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} formData = {'name': '', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'type': volume_type.name, 'size': 50, 'snapshot_source': '', 'availability_zone': az} cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) cinder.volume_create(IsA(http.HttpRequest), formData['size'], formData['name'], formData['description'], formData['type'], metadata={}, snapshot_id=None, image_id=None, availability_zone=formData['availability_zone'], source_volid=None)\ .AndReturn(volume) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') res = self.client.post(url, formData) redirect_url = VOLUME_VOLUMES_TAB_URL self.assertRedirectsNoFollow(res, redirect_url) @test.create_stubs({cinder: ('volume_create', 'volume_snapshot_list', 'volume_type_list', 'volume_type_default', 'volume_list', 'availability_zone_list', 'extension_supported'), api.glance: ('image_list_detailed',), quotas: ('tenant_limit_usages',)}) def test_create_volume_dropdown(self): volume = self.cinder_volumes.first() usage_limit = {'maxTotalVolumeGigabytes': 250, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 50, 'type': '', 'volume_source_type': 'no_source_type', 'snapshot_source': self.cinder_volume_snapshots.first().id, 'image_source': self.images.first().id} cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) cinder.volume_create(IsA(http.HttpRequest), formData['size'], formData['name'], formData['description'], '', metadata={}, snapshot_id=None, image_id=None, availability_zone=None, source_volid=None).AndReturn(volume) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') res = self.client.post(url, formData) redirect_url = VOLUME_VOLUMES_TAB_URL self.assertRedirectsNoFollow(res, redirect_url) @test.create_stubs({cinder: ('volume_create', 'volume_snapshot_get', 'volume_get', 'volume_type_default', 'volume_type_list'), quotas: ('tenant_limit_usages',)}) def test_create_volume_from_snapshot(self): volume = self.cinder_volumes.first() usage_limit = {'maxTotalVolumeGigabytes': 250, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} snapshot = self.cinder_volume_snapshots.first() formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 50, 'type': '', 'snapshot_source': snapshot.id} cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) cinder.volume_snapshot_get(IsA(http.HttpRequest), str(snapshot.id)).AndReturn(snapshot) cinder.volume_get(IsA(http.HttpRequest), snapshot.volume_id).\ AndReturn(self.cinder_volumes.first()) cinder.volume_create(IsA(http.HttpRequest), formData['size'], formData['name'], formData['description'], '', metadata={}, snapshot_id=snapshot.id, image_id=None, availability_zone=None, source_volid=None).AndReturn(volume) self.mox.ReplayAll() # get snapshot from url url = reverse('horizon:project:volumes:volumes:create') res = self.client.post("?".join([url, "snapshot_id=" + str(snapshot.id)]), formData) redirect_url = VOLUME_VOLUMES_TAB_URL self.assertRedirectsNoFollow(res, redirect_url) @test.create_stubs({cinder: ('volume_create', 'volume_get', 'volume_type_default', 'volume_list', 'volume_type_list', 'availability_zone_list', 'volume_snapshot_get', 'volume_snapshot_list', 'extension_supported'), api.glance: ('image_list_detailed',), quotas: ('tenant_limit_usages',)}) def test_create_volume_from_volume(self): volume = self.cinder_volumes.first() usage_limit = {'maxTotalVolumeGigabytes': 250, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} formData = {'name': u'A copy of a volume', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 50, 'type': '', 'volume_source_type': 'volume_source', 'volume_source': volume.id} cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volumes.list()) cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volume_snapshots.list()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(self.cinder_volumes.first()) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones').AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.volume_create(IsA(http.HttpRequest), formData['size'], formData['name'], formData['description'], '', metadata={}, snapshot_id=None, image_id=None, availability_zone=None, source_volid=volume.id).AndReturn(volume) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') redirect_url = VOLUME_VOLUMES_TAB_URL res = self.client.post(url, formData) self.assertNoFormErrors(res) self.assertMessageCount(info=1) self.assertRedirectsNoFollow(res, redirect_url) @test.create_stubs({cinder: ('volume_create', 'volume_snapshot_list', 'volume_snapshot_get', 'volume_get', 'volume_type_default', 'volume_list', 'volume_type_list', 'availability_zone_list', 'extension_supported'), api.glance: ('image_list_detailed',), quotas: ('tenant_limit_usages',)}) def test_create_volume_from_snapshot_dropdown(self): volume = self.cinder_volumes.first() usage_limit = {'maxTotalVolumeGigabytes': 250, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} snapshot = self.cinder_volume_snapshots.first() formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 50, 'type': '', 'volume_source_type': 'snapshot_source', 'snapshot_source': snapshot.id} cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) cinder.volume_snapshot_get(IsA(http.HttpRequest), str(snapshot.id)).AndReturn(snapshot) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) cinder.volume_create(IsA(http.HttpRequest), formData['size'], formData['name'], formData['description'], '', metadata={}, snapshot_id=snapshot.id, image_id=None, availability_zone=None, source_volid=None).AndReturn(volume) self.mox.ReplayAll() # get snapshot from dropdown list url = reverse('horizon:project:volumes:volumes:create') res = self.client.post(url, formData) redirect_url = VOLUME_VOLUMES_TAB_URL self.assertRedirectsNoFollow(res, redirect_url) @test.create_stubs({cinder: ('volume_snapshot_get', 'volume_type_list', 'volume_type_default', 'volume_get'), api.glance: ('image_list_detailed',), quotas: ('tenant_limit_usages',)}) def test_create_volume_from_snapshot_invalid_size(self): usage_limit = {'maxTotalVolumeGigabytes': 100, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} snapshot = self.cinder_volume_snapshots.first() formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 20, 'snapshot_source': snapshot.id} cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) if django.VERSION >= (1, 9): cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).MultipleTimes().\ AndReturn(usage_limit) cinder.volume_snapshot_get(IsA(http.HttpRequest), str(snapshot.id)).AndReturn(snapshot) cinder.volume_get(IsA(http.HttpRequest), snapshot.volume_id).\ AndReturn(self.cinder_volumes.first()) if django.VERSION >= (1, 9): cinder.volume_type_default(IsA(http.HttpRequest)). \ AndReturn(self.cinder_volume_types.first()) cinder.volume_snapshot_get(IsA(http.HttpRequest), str(snapshot.id)).AndReturn(snapshot) cinder.volume_get(IsA(http.HttpRequest), snapshot.volume_id). \ AndReturn(self.cinder_volumes.first()) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') res = self.client.post("?".join([url, "snapshot_id=" + str(snapshot.id)]), formData, follow=True) self.assertEqual(res.redirect_chain, []) self.assertFormError(res, 'form', None, "The volume size cannot be less than the " "snapshot size (40GiB)") @test.create_stubs({cinder: ('volume_create', 'volume_type_list', 'volume_type_default', 'availability_zone_list', 'extension_supported'), api.glance: ('image_get',), quotas: ('tenant_limit_usages',)}) def test_create_volume_from_image(self): volume = self.cinder_volumes.first() usage_limit = {'maxTotalVolumeGigabytes': 200, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} image = self.images.first() formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 40, 'type': '', 'image_source': image.id} cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) api.glance.image_get(IsA(http.HttpRequest), str(image.id)).AndReturn(image) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) cinder.volume_create(IsA(http.HttpRequest), formData['size'], formData['name'], formData['description'], '', metadata={}, snapshot_id=None, image_id=image.id, availability_zone=None, source_volid=None).AndReturn(volume) self.mox.ReplayAll() # get image from url url = reverse('horizon:project:volumes:volumes:create') res = self.client.post("?".join([url, "image_id=" + str(image.id)]), formData) redirect_url = VOLUME_VOLUMES_TAB_URL self.assertRedirectsNoFollow(res, redirect_url) @test.create_stubs({cinder: ('volume_create', 'volume_type_list', 'volume_type_default', 'volume_list', 'volume_snapshot_list', 'availability_zone_list', 'extension_supported'), api.glance: ('image_get', 'image_list_detailed'), quotas: ('tenant_limit_usages',)}) def test_create_volume_from_image_dropdown(self): volume = self.cinder_volumes.first() usage_limit = {'maxTotalVolumeGigabytes': 200, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} image = self.images.first() formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 30, 'type': '', 'volume_source_type': 'image_source', 'snapshot_source': self.cinder_volume_snapshots.first().id, 'image_source': image.id} cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) quotas.tenant_limit_usages(IsA(http.HttpRequest)) \ .AndReturn(usage_limit) api.glance.image_get(IsA(http.HttpRequest), str(image.id)).AndReturn(image) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) cinder.volume_create(IsA(http.HttpRequest), formData['size'], formData['name'], formData['description'], '', metadata={}, snapshot_id=None, image_id=image.id, availability_zone=None, source_volid=None).AndReturn(volume) self.mox.ReplayAll() # get image from dropdown list url = reverse('horizon:project:volumes:volumes:create') res = self.client.post(url, formData) redirect_url = VOLUME_VOLUMES_TAB_URL self.assertRedirectsNoFollow(res, redirect_url) @test.create_stubs({cinder: ('volume_type_list', 'volume_type_default', 'availability_zone_list', 'extension_supported'), api.glance: ('image_get', 'image_list_detailed'), quotas: ('tenant_limit_usages',)}) def test_create_volume_from_image_under_image_size(self): usage_limit = {'maxTotalVolumeGigabytes': 100, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} image = self.images.first() formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 1, 'image_source': image.id} cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) if django.VERSION >= (1, 9): cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).MultipleTimes().\ AndReturn(usage_limit) api.glance.image_get(IsA(http.HttpRequest), str(image.id)).AndReturn(image) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) if django.VERSION >= (1, 9): cinder.volume_type_default(IsA(http.HttpRequest)). \ AndReturn(self.cinder_volume_types.first()) api.glance.image_get(IsA(http.HttpRequest), str(image.id)).AndReturn(image) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones').AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') res = self.client.post("?".join([url, "image_id=" + str(image.id)]), formData, follow=True) self.assertEqual(res.redirect_chain, []) msg = (u"The volume size cannot be less than the " u"image size (20.0\xa0GB)") self.assertFormError(res, 'form', None, msg) @test.create_stubs({cinder: ('volume_type_list', 'volume_type_default', 'availability_zone_list', 'extension_supported'), api.glance: ('image_get', 'image_list_detailed'), quotas: ('tenant_limit_usages',)}) def _test_create_volume_from_image_under_image_min_disk_size(self, image): usage_limit = {'maxTotalVolumeGigabytes': 100, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} formData = {'name': u'A Volume I Am Making', 'description': u'This is a volume I am making for a test.', 'method': u'CreateForm', 'size': 5, 'image_source': image.id} cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) if django.VERSION >= (1, 9): cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).MultipleTimes().\ AndReturn(usage_limit) api.glance.image_get(IsA(http.HttpRequest), str(image.id)).AndReturn(image) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) if django.VERSION >= (1, 9): cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) api.glance.image_get(IsA(http.HttpRequest), str(image.id)).AndReturn(image) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones').AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') res = self.client.post("?".join([url, "image_id=" + str(image.id)]), formData, follow=True) self.assertEqual(res.redirect_chain, []) self.assertFormError(res, 'form', None, "The volume size cannot be less than the " "image minimum disk size (30GiB)") def test_create_volume_from_image_under_image_min_disk_size(self): image = self.images.get(name="protected_images") image.min_disk = 30 self._test_create_volume_from_image_under_image_min_disk_size(image) @override_settings(OPENSTACK_API_VERSIONS={'image': 1}) def test_create_volume_from_image_under_image_prop_min_disk_size_v1(self): image = self.images.get(name="protected_images") image.min_disk = 0 image.properties['min_disk'] = 30 self._test_create_volume_from_image_under_image_min_disk_size(image) def test_create_volume_from_image_under_image_prop_min_disk_size_v2(self): image = self.imagesV2.get(name="protected_images") self._test_create_volume_from_image_under_image_min_disk_size(image) @test.create_stubs({cinder: ('volume_snapshot_list', 'volume_type_list', 'volume_type_default', 'volume_list', 'availability_zone_list', 'extension_supported'), api.glance: ('image_list_detailed',), quotas: ('tenant_limit_usages',)}) def test_create_volume_gb_used_over_alloted_quota(self): usage_limit = {'maxTotalVolumeGigabytes': 100, 'gigabytesUsed': 80, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': 6} formData = {'name': u'This Volume Is Huge!', 'description': u'This is a volume that is just too big!', 'method': u'CreateForm', 'size': 5000} cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) if django.VERSION >= (1, 9): cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).MultipleTimes().\ AndReturn(usage_limit) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) if django.VERSION >= (1, 9): cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS). \ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones') \ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') res = self.client.post(url, formData) expected_error = [u'A volume of 5000GiB cannot be created as you only' ' have 20GiB of your quota available.'] self.assertEqual(res.context['form'].errors['__all__'], expected_error) @test.create_stubs({cinder: ('volume_snapshot_list', 'volume_type_default', 'volume_type_list', 'volume_list', 'availability_zone_list', 'extension_supported'), api.glance: ('image_list_detailed',), quotas: ('tenant_limit_usages',)}) def test_create_volume_number_over_alloted_quota(self): usage_limit = {'maxTotalVolumeGigabytes': 100, 'gigabytesUsed': 20, 'volumesUsed': len(self.cinder_volumes.list()), 'maxTotalVolumes': len(self.cinder_volumes.list())} formData = {'name': u'Too Many...', 'description': u'We have no volumes left!', 'method': u'CreateForm', 'size': 10} cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) if django.VERSION >= (1, 9): cinder.volume_type_list(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.list()) cinder.volume_type_default(IsA(http.HttpRequest)).\ AndReturn(self.cinder_volume_types.first()) quotas.tenant_limit_usages(IsA(http.HttpRequest)).MultipleTimes().\ AndReturn(usage_limit) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS).\ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones')\ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) if django.VERSION >= (1, 9): cinder.volume_type_default(IsA(http.HttpRequest)). \ AndReturn(self.cinder_volume_types.first()) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=SEARCH_OPTS). \ AndReturn(self.cinder_volume_snapshots.list()) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'is_public': True, 'status': 'active'}) \ .AndReturn([self.images.list(), False, False]) api.glance.image_list_detailed( IsA(http.HttpRequest), filters={'property-owner_id': self.tenant.id, 'status': 'active'}) \ .AndReturn([[], False, False]) cinder.volume_list(IsA( http.HttpRequest), search_opts=SEARCH_OPTS).AndReturn(self.cinder_volumes.list()) cinder.extension_supported(IsA(http.HttpRequest), 'AvailabilityZones') \ .AndReturn(True) cinder.availability_zone_list(IsA(http.HttpRequest)).AndReturn( self.cinder_availability_zones.list()) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:create') res = self.client.post(url, formData) expected_error = [u'You are already using all of your available' ' volumes.'] self.assertEqual(res.context['form'].errors['__all__'], expected_error) @test.create_stubs({cinder: ('tenant_absolute_limits', 'volume_list_paged', 'volume_snapshot_list', 'volume_backup_supported', 'volume_delete',), api.nova: ('server_list',)}) def test_delete_volume(self): volumes = self.cinder_volumes.list() volume = self.cinder_volumes.first() formData = {'action': 'volumes__delete__%s' % volume.id} cinder.volume_backup_supported(IsA(http.HttpRequest)). \ MultipleTimes().AndReturn(True) cinder.volume_list_paged( IsA(http.HttpRequest), marker=None, paginate=True, sort_dir='desc', search_opts=None).AndReturn([volumes, False, False]) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=None).\ AndReturn([]) cinder.volume_delete(IsA(http.HttpRequest), volume.id) api.nova.server_list(IsA(http.HttpRequest), search_opts=None).\ AndReturn([self.servers.list(), False]) cinder.volume_list_paged( IsA(http.HttpRequest), marker=None, paginate=True, sort_dir='desc', search_opts=None).AndReturn([volumes, False, False]) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=None).\ AndReturn([]) api.nova.server_list(IsA(http.HttpRequest), search_opts=None).\ AndReturn([self.servers.list(), False]) cinder.tenant_absolute_limits(IsA(http.HttpRequest)).MultipleTimes().\ AndReturn(self.cinder_limits['absolute']) self.mox.ReplayAll() url = VOLUME_INDEX_URL res = self.client.post(url, formData, follow=True) self.assertIn("Scheduled deletion of Volume: Volume name", [m.message for m in res.context['messages']]) @test.create_stubs({cinder: ('volume_get', 'tenant_absolute_limits')}) def test_delete_volume_with_snap_no_action_item(self): volume = self.cinder_volumes.get(name='Volume name') setattr(volume, 'has_snapshot', True) limits = self.cinder_limits['absolute'] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.tenant_absolute_limits(IsA(http.HttpRequest)). \ MultipleTimes('limits').AndReturn(limits) self.mox.ReplayAll() url = (VOLUME_INDEX_URL + "?action=row_update&table=volumes&obj_id=" + volume.id) res = self.client.get(url, {}, HTTP_X_REQUESTED_WITH='XMLHttpRequest') self.assertEqual(res.status_code, 200) self.assertNotContains(res, 'Delete Volume') self.assertNotContains(res, 'delete') @test.create_stubs({cinder: ('volume_get',), api.nova: ('server_list',)}) @override_settings(OPENSTACK_HYPERVISOR_FEATURES={'can_set_mount_point': True}) def test_edit_attachments(self): volume = self.cinder_volumes.first() servers = [s for s in self.servers.list() if s.tenant_id == self.request.user.tenant_id] volume.attachments = [{'id': volume.id, 'volume_id': volume.id, 'volume_name': volume.name, 'instance': servers[0], 'device': '/dev/vdb', 'server_id': servers[0].id}] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) api.nova.server_list(IsA(http.HttpRequest)).AndReturn([servers, False]) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:attach', args=[volume.id]) res = self.client.get(url) msg = 'Volume %s on instance %s' % (volume.name, servers[0].name) self.assertContains(res, msg) # Asserting length of 2 accounts for the one instance option, # and the one 'Choose Instance' option. form = res.context['form'] self.assertEqual(len(form.fields['instance']._choices), 1) self.assertEqual(res.status_code, 200) self.assertIsInstance(form.fields['device'].widget, widgets.TextInput) self.assertFalse(form.fields['device'].required) @test.create_stubs({cinder: ('volume_get',), api.nova: ('server_list',)}) @override_settings(OPENSTACK_HYPERVISOR_FEATURES={'can_set_mount_point': True}) def test_edit_attachments_auto_device_name(self): volume = self.cinder_volumes.first() servers = [s for s in self.servers.list() if s.tenant_id == self.request.user.tenant_id] volume.attachments = [{'id': volume.id, 'volume_id': volume.id, 'volume_name': volume.name, 'instance': servers[0], 'device': '', 'server_id': servers[0].id}] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) api.nova.server_list(IsA(http.HttpRequest)).AndReturn([servers, False]) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:attach', args=[volume.id]) res = self.client.get(url) form = res.context['form'] self.assertIsInstance(form.fields['device'].widget, widgets.TextInput) self.assertFalse(form.fields['device'].required) @test.create_stubs({cinder: ('volume_get',), api.nova: ('server_list',)}) def test_edit_attachments_cannot_set_mount_point(self): volume = self.cinder_volumes.first() servers = [s for s in self.servers.list() if s.tenant_id == self.request.user.tenant_id] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) api.nova.server_list(IsA(http.HttpRequest)).AndReturn([servers, False]) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:attach', args=[volume.id]) res = self.client.get(url) # Assert the device field is hidden. form = res.context['form'] self.assertIsInstance(form.fields['device'].widget, widgets.HiddenInput) @test.create_stubs({cinder: ('volume_get',), api.nova: ('server_list',)}) def test_edit_attachments_attached_volume(self): servers = [s for s in self.servers.list() if s.tenant_id == self.request.user.tenant_id] server = servers[0] volume = self.cinder_volumes.list()[0] cinder.volume_get(IsA(http.HttpRequest), volume.id) \ .AndReturn(volume) api.nova.server_list(IsA(http.HttpRequest)) \ .AndReturn([servers, False]) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:attach', args=[volume.id]) res = self.client.get(url) self.assertEqual(res.context['form'].fields['instance']._choices[0][1], "Select an instance") self.assertEqual(len(res.context['form'].fields['instance'].choices), 2) self.assertEqual(res.context['form'].fields['instance']._choices[1][0], server.id) self.assertEqual(res.status_code, 200) def _get_volume_row_action_from_ajax(self, res, action_name, row_id): def _matches_row_id(context_row): return (len(context_row.dicts) > 1 and hasattr(context_row.dicts[1], 'get') and context_row.dicts[1].get('row_id', None) == row_id) matching = list(moves.filter(lambda r: _matches_row_id(r), res.context)) self.assertTrue(len(matching) > 1, "Expected at least one row matching %s" % row_id) row = matching[-1].dicts[1] matching_actions = list(moves.filter(lambda a: a.name == action_name, row['row_actions'])) self.assertEqual(1, len(matching_actions), "Expected one row action named '%s'" % action_name) return matching_actions[0] @test.create_stubs({cinder: ('tenant_absolute_limits', 'volume_get',)}) def test_create_snapshot_button_attributes(self): limits = {'maxTotalSnapshots': 2} limits['totalSnapshotsUsed'] = 1 volume = self.cinder_volumes.first() cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.tenant_absolute_limits(IsA(http.HttpRequest)).AndReturn(limits) self.mox.ReplayAll() res_url = (VOLUME_INDEX_URL + "?action=row_update&table=volumes&obj_id=" + volume.id) res = self.client.get(res_url, {}, HTTP_X_REQUESTED_WITH='XMLHttpRequest') snapshot_action = self._get_volume_row_action_from_ajax( res, 'snapshots', volume.id) self.assertEqual('horizon:project:volumes:volumes:create_snapshot', snapshot_action.url) self.assertEqual(set(['ajax-modal']), set(snapshot_action.classes)) self.assertEqual('Create Snapshot', six.text_type(snapshot_action.verbose_name)) self.assertEqual((('volume', 'volume:create_snapshot'),), snapshot_action.policy_rules) @test.create_stubs({cinder: ('tenant_absolute_limits', 'volume_get',)}) def test_create_snapshot_button_disabled_when_quota_exceeded(self): limits = {'maxTotalSnapshots': 1} limits['totalSnapshotsUsed'] = limits['maxTotalSnapshots'] volume = self.cinder_volumes.first() cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.tenant_absolute_limits(IsA(http.HttpRequest)).AndReturn(limits) self.mox.ReplayAll() res_url = (VOLUME_INDEX_URL + "?action=row_update&table=volumes&obj_id=" + volume.id) res = self.client.get(res_url, {}, HTTP_X_REQUESTED_WITH='XMLHttpRequest') snapshot_action = self._get_volume_row_action_from_ajax( res, 'snapshots', volume.id) self.assertTrue('disabled' in snapshot_action.classes, 'The create snapshot button should be disabled') @test.create_stubs({cinder: ('tenant_absolute_limits', 'volume_list_paged', 'volume_snapshot_list', 'volume_backup_supported',), api.nova: ('server_list',)}) def test_create_button_attributes(self): limits = self.cinder_limits['absolute'] limits['maxTotalVolumes'] = 10 limits['totalVolumesUsed'] = 1 volumes = self.cinder_volumes.list() api.cinder.volume_backup_supported(IsA(http.HttpRequest)). \ MultipleTimes().AndReturn(True) cinder.volume_list_paged(IsA(http.HttpRequest), sort_dir='desc', marker=None, paginate=True, search_opts=None)\ .AndReturn([volumes, False, False]) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=None).\ AndReturn([]) api.nova.server_list(IsA(http.HttpRequest), search_opts=None)\ .AndReturn([self.servers.list(), False]) cinder.tenant_absolute_limits(IsA(http.HttpRequest))\ .MultipleTimes().AndReturn(limits) self.mox.ReplayAll() res = self.client.get(VOLUME_INDEX_URL) self.assertTemplateUsed(res, 'project/volumes/index.html') volumes = res.context['volumes_table'].data self.assertItemsEqual(volumes, self.cinder_volumes.list()) create_action = self.getAndAssertTableAction(res, 'volumes', 'create') self.assertEqual(set(['ajax-modal', 'ajax-update', 'btn-create']), set(create_action.classes)) self.assertEqual('Create Volume', six.text_type(create_action.verbose_name)) self.assertEqual('horizon:project:volumes:volumes:create', create_action.url) self.assertEqual((('volume', 'volume:create'),), create_action.policy_rules) @test.create_stubs({cinder: ('tenant_absolute_limits', 'volume_list_paged', 'volume_snapshot_list', 'volume_backup_supported',), api.nova: ('server_list',)}) def test_create_button_disabled_when_quota_exceeded(self): limits = self.cinder_limits['absolute'] limits['totalVolumesUsed'] = limits['maxTotalVolumes'] volumes = self.cinder_volumes.list() api.cinder.volume_backup_supported(IsA(http.HttpRequest)). \ MultipleTimes().AndReturn(True) cinder.volume_list_paged( IsA(http.HttpRequest), marker=None, paginate=True, sort_dir='desc', search_opts=None).AndReturn([volumes, False, False]) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts=None).\ AndReturn([]) api.nova.server_list(IsA(http.HttpRequest), search_opts=None)\ .AndReturn([self.servers.list(), False]) cinder.tenant_absolute_limits(IsA(http.HttpRequest))\ .MultipleTimes().AndReturn(limits) self.mox.ReplayAll() res = self.client.get(VOLUME_INDEX_URL) self.assertTemplateUsed(res, 'project/volumes/index.html') volumes = res.context['volumes_table'].data self.assertItemsEqual(volumes, self.cinder_volumes.list()) create_action = self.getAndAssertTableAction(res, 'volumes', 'create') self.assertTrue('disabled' in create_action.classes, 'The create button should be disabled') @test.create_stubs({cinder: ('tenant_absolute_limits', 'volume_get', 'volume_snapshot_list'), api.nova: ('server_get',)}) def test_detail_view(self): volume = self.cinder_volumes.first() server = self.servers.first() snapshots = self.cinder_volume_snapshots.list() volume.attachments = [{"server_id": server.id}] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts={'volume_id': volume.id})\ .AndReturn(snapshots) api.nova.server_get(IsA(http.HttpRequest), server.id).AndReturn(server) cinder.tenant_absolute_limits(IsA(http.HttpRequest))\ .AndReturn(self.cinder_limits['absolute']) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:detail', args=[volume.id]) res = self.client.get(url) self.assertTemplateUsed(res, 'horizon/common/_detail.html') self.assertEqual(res.context['volume'].id, volume.id) self.assertNoMessages() @test.create_stubs({cinder: ('volume_get', 'volume_get_encryption_metadata'), }) def test_encryption_detail_view_encrypted(self): enc_meta = self.cinder_volume_encryption.first() volume = self.cinder_volumes.get(name='my_volume2') cinder.volume_get_encryption_metadata( IsA(http.HttpRequest), volume.id).AndReturn(enc_meta) cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:encryption_detail', args=[volume.id]) res = self.client.get(url) self.assertContains(res, "Volume Encryption Details: %s" % volume.name, 2, 200) self.assertContains(res, "