e4444e6979
In Django 3.0, django.utils.http.urlquote(), urlquote_plus(), urlunquote(), and urlunquote_plus() are deprecated in favor of the functions that they’re aliases for: urllib.parse.quote(), quote_plus(), unquote(), and unquote_plus(). https: //docs.djangoproject.com/en/4.0/releases/3.0/ Change-Id: I37fcd917cbf87b4d3141cfbdd2675aa38f33f2a4
416 lines
17 KiB
Python
416 lines
17 KiB
Python
# 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 copy
|
|
from urllib import parse
|
|
|
|
from django.conf import settings
|
|
from django.test.utils import override_settings
|
|
from django.urls import reverse
|
|
|
|
from openstack_dashboard import api
|
|
from openstack_dashboard.dashboards.project.volumes \
|
|
import tables as volume_tables
|
|
from openstack_dashboard.test import helpers as test
|
|
|
|
|
|
DETAIL_URL = ('horizon:admin:volumes:detail')
|
|
INDEX_URL = reverse('horizon:admin:volumes:index')
|
|
|
|
|
|
class VolumeTests(test.BaseAdminViewTests):
|
|
def tearDown(self):
|
|
for volume in self.cinder_volumes.list():
|
|
# VolumeTableMixIn._set_volume_attributes mutates data
|
|
# and cinder_volumes.list() doesn't deep copy
|
|
for att in volume.attachments:
|
|
if 'instance' in att:
|
|
del att['instance']
|
|
super().tearDown()
|
|
|
|
@test.create_mocks({
|
|
api.nova: ['server_list'],
|
|
api.cinder: ['volume_snapshot_list', 'volume_list_paged'],
|
|
api.keystone: ['tenant_list']})
|
|
def _test_index(self, instanceless_volumes):
|
|
volumes = self.cinder_volumes.list()
|
|
if instanceless_volumes:
|
|
for volume in volumes:
|
|
volume.attachments = []
|
|
|
|
self.mock_volume_list_paged.return_value = [volumes, False, False]
|
|
self.mock_volume_snapshot_list.return_value = []
|
|
|
|
if not instanceless_volumes:
|
|
self.mock_server_list.return_value = [self.servers.list(), False]
|
|
|
|
self.mock_tenant_list.return_value = [[self.tenants.list(), False]]
|
|
|
|
res = self.client.get(INDEX_URL)
|
|
if not instanceless_volumes:
|
|
self.mock_server_list.assert_called_once_with(
|
|
test.IsHttpRequest(), search_opts={'all_tenants': True})
|
|
|
|
self.mock_volume_list_paged.assert_called_once_with(
|
|
test.IsHttpRequest(), sort_dir="desc", marker=None, paginate=True,
|
|
search_opts={'all_tenants': True})
|
|
self.mock_volume_snapshot_list.assert_called_once_with(
|
|
test.IsHttpRequest(), search_opts={'all_tenants': True})
|
|
self.mock_tenant_list.assert_called_once()
|
|
self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')
|
|
volumes = res.context['volumes_table'].data
|
|
self.assertCountEqual(volumes, self.cinder_volumes.list())
|
|
|
|
def test_index_without_attachments(self):
|
|
self._test_index(True)
|
|
|
|
def test_index_with_attachments(self):
|
|
self._test_index(False)
|
|
|
|
@test.create_mocks({
|
|
api.nova: ['server_list'],
|
|
api.cinder: ['volume_snapshot_list', 'volume_list_paged'],
|
|
api.keystone: ['tenant_list']})
|
|
def _test_index_paginated(self, marker, sort_dir, volumes, url,
|
|
has_more, has_prev):
|
|
vol_snaps = self.cinder_volume_snapshots.list()
|
|
|
|
self.mock_volume_list_paged.return_value = \
|
|
[volumes, has_more, has_prev]
|
|
self.mock_volume_snapshot_list.return_value = vol_snaps
|
|
self.mock_server_list.return_value = [self.servers.list(), False]
|
|
self.mock_tenant_list.return_value = [self.tenants.list(), False]
|
|
|
|
res = self.client.get(parse.unquote(url))
|
|
|
|
self.mock_server_list.assert_called_once_with(
|
|
test.IsHttpRequest(), search_opts={'all_tenants': True})
|
|
self.mock_volume_list_paged.assert_called_once_with(
|
|
test.IsHttpRequest(),
|
|
sort_dir=sort_dir,
|
|
marker=marker, paginate=True,
|
|
search_opts={
|
|
'all_tenants': True})
|
|
self.mock_volume_snapshot_list.assert_called_once_with(
|
|
test.IsHttpRequest(), search_opts={'all_tenants': True})
|
|
self.mock_tenant_list.assert_called_once()
|
|
|
|
self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
return res
|
|
|
|
@override_settings(FILTER_DATA_FIRST={'admin.volumes': True})
|
|
def test_volumes_tab_with_admin_filter_first(self):
|
|
res = self.client.get(INDEX_URL)
|
|
self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')
|
|
volumes = res.context['volumes_table'].data
|
|
self.assertCountEqual(volumes, [])
|
|
|
|
def _ensure_attachments_exist(self, volumes):
|
|
volumes = copy.copy(volumes)
|
|
for volume in volumes:
|
|
if not volume.attachments:
|
|
volume.attachments.append({
|
|
"id": "1", "server_id": '1', "device": "/dev/hda"})
|
|
return volumes
|
|
|
|
@override_settings(API_RESULT_PAGE_SIZE=2)
|
|
def test_index_paginated(self):
|
|
size = settings.API_RESULT_PAGE_SIZE
|
|
volumes = self._ensure_attachments_exist(
|
|
self.cinder_volumes.list())
|
|
|
|
# get first page
|
|
expected_volumes = volumes[:size]
|
|
url = INDEX_URL
|
|
res = self._test_index_paginated(None, "desc", expected_volumes, url,
|
|
True, False)
|
|
result = res.context['volumes_table'].data
|
|
self.assertCountEqual(result, expected_volumes)
|
|
|
|
# get second page
|
|
expected_volumes = volumes[size:2 * size]
|
|
marker = expected_volumes[0].id
|
|
next = volume_tables.VolumesTable._meta.pagination_param
|
|
url = INDEX_URL + "?%s=%s" % (next, marker)
|
|
res = self._test_index_paginated(marker, "desc", expected_volumes, url,
|
|
True, True)
|
|
result = res.context['volumes_table'].data
|
|
self.assertCountEqual(result, expected_volumes)
|
|
|
|
# get last page
|
|
expected_volumes = volumes[-size:]
|
|
marker = expected_volumes[0].id
|
|
next = volume_tables.VolumesTable._meta.pagination_param
|
|
url = INDEX_URL + "?%s=%s" % (next, marker)
|
|
res = self._test_index_paginated(marker, "desc", expected_volumes, url,
|
|
False, True)
|
|
result = res.context['volumes_table'].data
|
|
self.assertCountEqual(result, expected_volumes)
|
|
|
|
@override_settings(API_RESULT_PAGE_SIZE=2)
|
|
def test_index_paginated_prev(self):
|
|
size = settings.API_RESULT_PAGE_SIZE
|
|
volumes = self._ensure_attachments_exist(
|
|
self.cinder_volumes.list())
|
|
|
|
# prev from some page
|
|
expected_volumes = volumes[size:2 * size]
|
|
marker = volumes[0].id
|
|
prev = volume_tables.VolumesTable._meta.prev_pagination_param
|
|
url = INDEX_URL + "?%s=%s" % (prev, marker)
|
|
res = self._test_index_paginated(marker, "asc", expected_volumes, url,
|
|
False, True)
|
|
result = res.context['volumes_table'].data
|
|
self.assertCountEqual(result, expected_volumes)
|
|
|
|
# back to first page
|
|
expected_volumes = volumes[:size]
|
|
marker = volumes[0].id
|
|
prev = volume_tables.VolumesTable._meta.prev_pagination_param
|
|
url = INDEX_URL + "?%s=%s" % (prev, marker)
|
|
res = self._test_index_paginated(marker, "asc", expected_volumes, url,
|
|
True, False)
|
|
result = res.context['volumes_table'].data
|
|
self.assertCountEqual(result, expected_volumes)
|
|
|
|
@test.create_mocks({api.cinder: ['volume_get', 'volume_reset_state']})
|
|
def test_update_volume_status(self):
|
|
volume = self.cinder_volumes.first()
|
|
form_data = {'status': 'error'}
|
|
|
|
self.mock_volume_reset_state.return_value = None
|
|
self.mock_volume_get.return_value = volume
|
|
|
|
res = self.client.post(
|
|
reverse('horizon:admin:volumes:update_status',
|
|
args=(volume.id,)),
|
|
form_data)
|
|
|
|
self.mock_volume_reset_state.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id, form_data['status'])
|
|
self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),
|
|
volume.id)
|
|
self.assertNoFormErrors(res)
|
|
|
|
@test.create_mocks({
|
|
api.cinder: ['extension_supported', 'availability_zone_list',
|
|
'volume_type_list', 'volume_manage']})
|
|
def test_manage_volume(self):
|
|
metadata = {'key': 'k1',
|
|
'value': 'v1'}
|
|
form_data = {'host': 'host-1',
|
|
'identifier': 'vol-1',
|
|
'id_type': 'source-name',
|
|
'name': 'name-1',
|
|
'description': 'manage a volume',
|
|
'volume_type': 'vol_type_1',
|
|
'availability_zone': 'nova',
|
|
'metadata': metadata['key'] + '=' + metadata['value'],
|
|
'bootable': False}
|
|
|
|
self.mock_extension_supported.return_value = None
|
|
self.mock_volume_type_list.return_value = \
|
|
self.cinder_volume_types.list()
|
|
self.mock_availability_zone_list.return_value = \
|
|
self.availability_zones.list()
|
|
self.mock_extension_supported.return_value = True
|
|
|
|
res = self.client.post(
|
|
reverse('horizon:admin:volumes:manage'),
|
|
form_data)
|
|
|
|
self.mock_volume_manage.assert_called_once_with(
|
|
test.IsHttpRequest(),
|
|
host=form_data['host'],
|
|
identifier=form_data['identifier'],
|
|
id_type=form_data['id_type'],
|
|
name=form_data['name'],
|
|
description=form_data['description'],
|
|
volume_type=form_data['volume_type'],
|
|
availability_zone=form_data['availability_zone'],
|
|
metadata={metadata['key']: metadata['value']},
|
|
bootable=form_data['bootable'])
|
|
self.mock_volume_type_list.assert_called_once()
|
|
self.mock_availability_zone_list.assert_called_once()
|
|
self.mock_extension_supported.assert_called_once_with(
|
|
test.IsHttpRequest(),
|
|
'AvailabilityZones')
|
|
self.assertNoFormErrors(res)
|
|
|
|
@test.create_mocks({api.cinder: ['volume_get', 'volume_unmanage']})
|
|
def test_unmanage_volume(self):
|
|
# important - need to get the v3 cinder volume which has host data
|
|
volume_list = [x for x in self.cinder_volumes.list()
|
|
if x.name == 'v3_volume']
|
|
volume = volume_list[0]
|
|
form_data = {'volume_name': volume.name,
|
|
'host_name': 'host@backend-name#pool',
|
|
'volume_id': volume.id}
|
|
|
|
self.mock_volume_get.return_value = volume
|
|
self.mock_volume_unmanage.return_value = volume
|
|
|
|
res = self.client.post(
|
|
reverse('horizon:admin:volumes:unmanage',
|
|
args=(volume.id,)),
|
|
form_data)
|
|
|
|
self.mock_volume_unmanage.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id)
|
|
self.mock_volume_get.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id)
|
|
self.assertNoFormErrors(res)
|
|
|
|
@test.create_mocks({api.cinder: ['volume_get', 'pool_list']})
|
|
def test_volume_migrate_get(self):
|
|
volume = self.cinder_volumes.get(name='v3_volume')
|
|
|
|
self.mock_pool_list.return_value = self.cinder_pools.list()
|
|
self.mock_volume_get.return_value = volume
|
|
|
|
url = reverse('horizon:admin:volumes:migrate',
|
|
args=[volume.id])
|
|
res = self.client.get(url)
|
|
|
|
self.mock_volume_get.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id)
|
|
self.mock_pool_list.assert_called_once()
|
|
self.assertTemplateUsed(res,
|
|
'admin/volumes/migrate_volume.html')
|
|
|
|
@test.create_mocks({api.cinder: ['volume_get']})
|
|
def test_volume_migrate_get_volume_get_exception(self):
|
|
volume = self.cinder_volumes.get(name='v3_volume')
|
|
self.mock_volume_get.side_effect = self.exceptions.cinder
|
|
|
|
url = reverse('horizon:admin:volumes:migrate',
|
|
args=[volume.id])
|
|
res = self.client.get(url)
|
|
|
|
self.mock_volume_get.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id)
|
|
self.assertRedirectsNoFollow(res, INDEX_URL)
|
|
|
|
@test.create_mocks({api.cinder: ['volume_get', 'pool_list']})
|
|
def test_volume_migrate_list_pool_get_exception(self):
|
|
volume = self.cinder_volumes.get(name='v3_volume')
|
|
|
|
self.mock_volume_get.return_value = volume
|
|
self.mock_pool_list.side_effect = self.exceptions.cinder
|
|
|
|
url = reverse('horizon:admin:volumes:migrate',
|
|
args=[volume.id])
|
|
res = self.client.get(url)
|
|
|
|
self.mock_volume_get.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id)
|
|
self.mock_pool_list.assert_called_once()
|
|
self.assertRedirectsNoFollow(res, INDEX_URL)
|
|
|
|
@test.create_mocks({
|
|
api.cinder: ['volume_migrate', 'volume_get', 'pool_list']})
|
|
def test_volume_migrate_post(self):
|
|
volume = self.cinder_volumes.get(name='v3_volume')
|
|
host = self.cinder_pools.first().name
|
|
|
|
self.mock_volume_get.return_value = volume
|
|
self.mock_pool_list.return_value = self.cinder_pools.list()
|
|
self.mock_volume_migrate.return_value = None
|
|
|
|
url = reverse('horizon:admin:volumes:migrate',
|
|
args=[volume.id])
|
|
res = self.client.post(url, {'host': host, 'volume_id': volume.id})
|
|
|
|
self.mock_volume_get.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id)
|
|
self.mock_pool_list.assert_called_once()
|
|
self.mock_volume_migrate.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id, host, False)
|
|
self.assertNoFormErrors(res)
|
|
self.assertRedirectsNoFollow(res, INDEX_URL)
|
|
|
|
@test.create_mocks({
|
|
api.cinder: ['volume_migrate', 'volume_get', 'pool_list']})
|
|
def test_volume_migrate_post_api_exception(self):
|
|
volume = self.cinder_volumes.get(name='v3_volume')
|
|
host = self.cinder_pools.first().name
|
|
|
|
self.mock_volume_get.return_value = volume
|
|
self.mock_pool_list.return_value = self.cinder_pools.list()
|
|
self.mock_volume_migrate.side_effect = self.exceptions.cinder
|
|
|
|
url = reverse('horizon:admin:volumes:migrate',
|
|
args=[volume.id])
|
|
res = self.client.post(url, {'host': host, 'volume_id': volume.id})
|
|
|
|
self.mock_volume_get.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id)
|
|
self.mock_pool_list.assert_called_once()
|
|
self.mock_volume_migrate.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id, host, False)
|
|
self.assertRedirectsNoFollow(res, INDEX_URL)
|
|
|
|
@test.create_mocks({api.cinder: ['volume_get']})
|
|
def test_update_volume_status_get(self):
|
|
volume = self.cinder_volumes.get(name='v3_volume')
|
|
self.mock_volume_get.return_value = volume
|
|
|
|
url = reverse('horizon:admin:volumes:update_status',
|
|
args=[volume.id])
|
|
res = self.client.get(url)
|
|
status_option = "<option value=\"%s\"></option>" % volume.status
|
|
|
|
self.mock_volume_get.assert_called_once_with(
|
|
test.IsHttpRequest(), volume.id)
|
|
self.assertNotContains(res, status_option)
|
|
|
|
@test.create_mocks({
|
|
api.nova: ['server_get'],
|
|
api.cinder: ['tenant_absolute_limits', 'volume_get',
|
|
'volume_snapshot_list']})
|
|
def test_detail_view_snapshot_tab(self):
|
|
volume = self.cinder_volumes.first()
|
|
server = self.servers.first()
|
|
snapshots = self.cinder_volume_snapshots.list()
|
|
this_volume_snapshots = [snapshot for snapshot in snapshots
|
|
if snapshot.volume_id == volume.id]
|
|
volume.attachments = [{"server_id": server.id}]
|
|
volume_limits = self.cinder_limits['absolute']
|
|
|
|
self.mock_server_get.return_value = server
|
|
self.mock_tenant_absolute_limits.return_value = volume_limits
|
|
self.mock_volume_get.return_value = volume
|
|
self.mock_volume_snapshot_list.return_value = this_volume_snapshots
|
|
|
|
url = (reverse(DETAIL_URL, args=[volume.id]) + '?' +
|
|
'='.join(['tab', 'volume_details__snapshots_tab']))
|
|
res = self.client.get(url)
|
|
|
|
self.assertTemplateUsed(res, 'horizon/common/_detail.html')
|
|
self.assertEqual(res.context['volume'].id, volume.id)
|
|
self.assertEqual(len(res.context['table'].data),
|
|
len(this_volume_snapshots))
|
|
self.assertNoMessages()
|
|
|
|
self.mock_server_get.assert_called_once_with(test.IsHttpRequest(),
|
|
server.id)
|
|
self.mock_tenant_absolute_limits.assert_called_once()
|
|
self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),
|
|
volume.id)
|
|
self.mock_volume_snapshot_list.assert_called_once_with(
|
|
test.IsHttpRequest(),
|
|
search_opts={'volume_id': volume.id, 'all_tenants': True})
|