openstack
/
horizon
Code Issues Proposed changes

Convert project tests into mock: api_access/backups/overview/snapshots 

Partially-Implements: blueprint mock-framework-in-unit-tests
Change-Id: I5a61d26d21f14ec44e8e82f6539196d53b6783fd
This commit is contained in:
Akihiro Motoki 2018-01-08 05:21:49 +09:00
parent d4c1c29cbb
commit dd560d95fd
4 changed files with 393 additions and 296 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
openstack_dashboard/dashboards/project/api_access/tests.py
View File

@ -12,11 +12,9 @@
# License for the specific language governing permissions and limitations
# under the License.
from mox3.mox import IsA
import six
import yaml
from django.http import HttpRequest
from django import template
from django.template import loader
from django.test.utils import override_settings
@ -36,23 +34,22 @@ RECREATE_CREDS_URL = reverse(API_URL + ":recreate_credentials")
class APIAccessTests(test.TestCase):
@test.create_mocks({api.keystone: ('create_ec2_credentials',
'list_ec2_credentials')})
def test_ec2_download_view(self):
creds = self.ec2.first()
self.mox.StubOutWithMock(api.keystone, "list_ec2_credentials")
self.mox.StubOutWithMock(api.keystone, "create_ec2_credentials")
api.keystone.list_ec2_credentials(IsA(HttpRequest), self.user.id) \
.AndReturn([])
api.keystone.create_ec2_credentials(IsA(HttpRequest),
self.user.id,
self.tenant.id).AndReturn(creds)
self.mox.ReplayAll()
self.mock_list_ec2_credentials.return_value = []
self.mock_create_ec2_credentials.return_value = creds
res = self.client.get(EC2_URL)
self.assertEqual(res.status_code, 200)
self.assertEqual(res['content-type'], 'application/zip')
self.mock_list_ec2_credentials.assert_called_once_with(
test.IsHttpRequest(), self.user.id)
self.mock_create_ec2_credentials.assert_called_once_with(
test.IsHttpRequest(), self.user.id, self.tenant.id)
def test_openrcv2_credentials(self):
res = self.client.get(OPENRCV2_URL)
self.assertEqual(res.status_code, 200)
@ -81,42 +78,32 @@ class APIAccessTests(test.TestCase):
self.assertIn(p_id.encode('utf-8'), res.content)
self.assertIn(domain.encode('utf-8'), res.content)
@test.create_stubs({api.keystone: ("list_ec2_credentials",)})
@test.create_mocks({api.keystone: ('list_ec2_credentials',)})
def test_credential_api(self):
certs = self.ec2.list()
api.keystone.list_ec2_credentials(IsA(HttpRequest), self.user.id) \
.AndReturn(certs)
self.mox.ReplayAll()
self.mock_list_ec2_credentials.return_value = certs
res = self.client.get(CREDS_URL)
self.assertEqual(res.status_code, 200)
credentials = 'project/api_access/credentials.html'
self.assertTemplateUsed(res, credentials)
self.assertEqual(self.user.id, res.context['openrc_creds']['user'].id)
self.assertEqual(certs[0].access,
res.context['ec2_creds']['ec2_access_key'])
self.mock_list_ec2_credentials.assert_called_once_with(
test.IsHttpRequest(), self.user.id)
@test.create_stubs({api.keystone: ("list_ec2_credentials",
"create_ec2_credentials",
"delete_user_ec2_credentials",)})
@test.create_mocks({api.keystone: ('create_ec2_credentials',
'list_ec2_credentials',
'delete_user_ec2_credentials')})
def _test_recreate_user_credentials(self, exists_credentials=True):
old_creds = self.ec2.list() if exists_credentials else []
new_creds = self.ec2.first()
api.keystone.list_ec2_credentials(
IsA(HttpRequest),
self.user.id).AndReturn(old_creds)
self.mock_list_ec2_credentials.return_value = old_creds
if exists_credentials:
api.keystone.delete_user_ec2_credentials(
IsA(HttpRequest),
self.user.id,
old_creds[0].access).AndReturn([])
api.keystone.create_ec2_credentials(
IsA(HttpRequest),
self.user.id,
self.tenant.id).AndReturn(new_creds)
self.mox.ReplayAll()
self.mock_delete_user_ec2_credentials.return_value = []
self.mock_create_ec2_credentials.return_value = new_creds
res_get = self.client.get(RECREATE_CREDS_URL)
self.assertEqual(res_get.status_code, 200)
@ -128,6 +115,16 @@ class APIAccessTests(test.TestCase):
self.assertNoFormErrors(res_post)
self.assertRedirectsNoFollow(res_post, INDEX_URL)
self.mock_list_ec2_credentials.assert_called_once_with(
test.IsHttpRequest(), self.user.id)
if exists_credentials:
self.mock_delete_user_ec2_credentials.assert_called_once_with(
test.IsHttpRequest(), self.user.id, old_creds[0].access)
else:
self.mock_delete_user_ec2_credentials.assert_not_called()
self.mock_create_ec2_credentials.assert_called_once_with(
test.IsHttpRequest(), self.user.id, self.tenant.id)
def test_recreate_user_credentials(self):
self._test_recreate_user_credentials()

144
openstack_dashboard/dashboards/project/backups/tests.py
View File

@ -11,12 +11,10 @@
# under the License.
from django.conf import settings
from django import http
from django.test.utils import override_settings
from django.urls import reverse
from django.utils.http import urlencode
from django.utils.http import urlunquote
from mox3.mox import IsA
from openstack_dashboard import api
from openstack_dashboard.dashboards.project.backups \
@ -29,24 +27,23 @@ INDEX_URL = reverse('horizon:project:backups:index')
class VolumeBackupsViewTests(test.TestCase):
@test.create_stubs({api.cinder: ('tenant_absolute_limits',
'volume_backup_list_paged',
'volume_list'),
api.nova: ('server_list',)})
@test.create_mocks({api.cinder: ('volume_list',
'volume_backup_list_paged')})
def _test_backups_index_paginated(self, marker, sort_dir, backups, url,
has_more, has_prev):
api.cinder.volume_backup_list_paged(
IsA(http.HttpRequest), marker=marker, sort_dir=sort_dir,
paginate=True).AndReturn([backups, has_more, has_prev])
api.cinder.volume_list(IsA(http.HttpRequest)).AndReturn(
self.cinder_volumes.list())
self.mox.ReplayAll()
self.mock_volume_backup_list_paged.return_value = [backups,
has_more, has_prev]
self.mock_volume_list.return_value = self.cinder_volumes.list()
res = self.client.get(urlunquote(url))
self.assertEqual(res.status_code, 200)
self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')
self.mock_volume_backup_list_paged.assert_called_once_with(
test.IsHttpRequest(), marker=marker, sort_dir=sort_dir,
paginate=True)
self.mock_volume_list.assert_called_once_with(test.IsHttpRequest())
self.mox.UnsetStubs()
return res
@override_settings(API_RESULT_PAGE_SIZE=1)
@ -112,21 +109,14 @@ class VolumeBackupsViewTests(test.TestCase):
backups = res.context['volume_backups_table'].data
self.assertItemsEqual(backups, expected_backups)
@test.create_stubs({api.cinder: ('volume_backup_create', 'volume_get')})
@test.create_mocks({api.cinder: ('volume_backup_create',
'volume_get')})
def test_create_backup_available(self):
volume = self.volumes.first()
backup = self.cinder_volume_backups.first()
api.cinder.volume_get(IsA(http.HttpRequest), volume.id). \
AndReturn(volume)
api.cinder.volume_backup_create(IsA(http.HttpRequest),
volume.id,
backup.container_name,
backup.name,
backup.description,
force=False) \
.AndReturn(backup)
self.mox.ReplayAll()
self.mock_volume_get.return_value = volume
self.mock_volume_backup_create.return_value = backup
formData = {'method': 'CreateBackupForm',
'tenant_id': self.tenant.id,
@ -142,22 +132,25 @@ class VolumeBackupsViewTests(test.TestCase):
self.assertMessageCount(error=0, warning=0)
self.assertRedirectsNoFollow(res, INDEX_URL)
@test.create_stubs({api.cinder: ('volume_backup_create', 'volume_get')})
self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),
volume.id)
self.mock_volume_backup_create.assert_called_once_with(
test.IsHttpRequest(),
volume.id,
backup.container_name,
backup.name,
backup.description,
force=False)
@test.create_mocks({api.cinder: ('volume_backup_create',
'volume_get')})
def test_create_backup_in_use(self):
# The second volume in the cinder test volume data is in-use
volume = self.volumes.list()[1]
backup = self.cinder_volume_backups.first()
api.cinder.volume_get(IsA(http.HttpRequest), volume.id). \
AndReturn(volume)
api.cinder.volume_backup_create(IsA(http.HttpRequest),
volume.id,
backup.container_name,
backup.name,
backup.description,
force=True) \
.AndReturn(backup)
self.mox.ReplayAll()
self.mock_volume_get.return_value = volume
self.mock_volume_backup_create.return_value = backup
formData = {'method': 'CreateBackupForm',
'tenant_id': self.tenant.id,
@ -172,8 +165,17 @@ class VolumeBackupsViewTests(test.TestCase):
self.assertNoFormErrors(res)
self.assertMessageCount(error=0, warning=0)
self.assertRedirectsNoFollow(res, INDEX_URL)
self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),
volume.id)
self.mock_volume_backup_create.assert_called_once_with(
test.IsHttpRequest(),
volume.id,
backup.container_name,
backup.name,
backup.description,
force=True)
@test.create_stubs({api.cinder: ('volume_list',
@test.create_mocks({api.cinder: ('volume_list',
'volume_backup_list_paged',
'volume_backup_delete')})
def test_delete_volume_backup(self):
@ -181,14 +183,10 @@ class VolumeBackupsViewTests(test.TestCase):
volumes = self.cinder_volumes.list()
backup = self.cinder_volume_backups.first()
api.cinder.volume_backup_list_paged(
IsA(http.HttpRequest), marker=None, sort_dir='desc',
paginate=True).AndReturn([vol_backups, False, False])
api.cinder.volume_list(IsA(http.HttpRequest)). \
AndReturn(volumes)
api.cinder.volume_backup_delete(IsA(http.HttpRequest), backup.id)
self.mox.ReplayAll()
self.mock_volume_backup_list_paged.return_value = [vol_backups,
False, False]
self.mock_volume_list.return_value = volumes
self.mock_volume_backup_delete.return_value = None
formData = {'action':
'volume_backups__delete__%s' % backup.id}
@ -196,17 +194,21 @@ class VolumeBackupsViewTests(test.TestCase):
self.assertRedirectsNoFollow(res, INDEX_URL)
self.assertMessageCount(success=1)
self.mock_volume_backup_list_paged.assert_called_once_with(
test.IsHttpRequest(), marker=None, sort_dir='desc',
paginate=True)
self.mock_volume_list.assert_called_once_with(test.IsHttpRequest())
self.mock_volume_backup_delete.assert_called_once_with(
test.IsHttpRequest(), backup.id)
@test.create_stubs({api.cinder: ('volume_backup_get', 'volume_get')})
@test.create_mocks({api.cinder: ('volume_backup_get',
'volume_get')})
def test_volume_backup_detail_get(self):
backup = self.cinder_volume_backups.first()
volume = self.cinder_volumes.get(id=backup.volume_id)
api.cinder.volume_backup_get(IsA(http.HttpRequest), backup.id). \
AndReturn(backup)
api.cinder.volume_get(IsA(http.HttpRequest), backup.volume_id). \
AndReturn(volume)
self.mox.ReplayAll()
self.mock_volume_backup_get.return_value = backup
self.mock_volume_get.return_value = volume
url = reverse('horizon:project:backups:detail',
args=[backup.id])
@ -214,15 +216,17 @@ class VolumeBackupsViewTests(test.TestCase):
self.assertTemplateUsed(res, 'horizon/common/_detail.html')
self.assertEqual(res.context['backup'].id, backup.id)
self.mock_volume_backup_get.assert_called_once_with(
test.IsHttpRequest(), backup.id)
self.mock_volume_get.assert_called_once_with(
test.IsHttpRequest(), backup.volume_id)
@test.create_stubs({api.cinder: ('volume_backup_get',)})
@test.create_mocks({api.cinder: ('volume_backup_get',)})
def test_volume_backup_detail_get_with_exception(self):
# Test to verify redirect if get volume backup fails
backup = self.cinder_volume_backups.first()
api.cinder.volume_backup_get(IsA(http.HttpRequest), backup.id).\
AndRaise(self.exceptions.cinder)
self.mox.ReplayAll()
self.mock_volume_backup_get.side_effect = self.exceptions.cinder
url = reverse('horizon:project:backups:detail',
args=[backup.id])
@ -231,17 +235,17 @@ class VolumeBackupsViewTests(test.TestCase):
self.assertNoFormErrors(res)
self.assertMessageCount(error=1)
self.assertRedirectsNoFollow(res, INDEX_URL)
self.mock_volume_backup_get.assert_called_once_with(
test.IsHttpRequest(), backup.id)
@test.create_stubs({api.cinder: ('volume_backup_get', 'volume_get')})
@test.create_mocks({api.cinder: ('volume_backup_get',
'volume_get')})
def test_volume_backup_detail_with_missing_volume(self):
# Test to check page still loads even if volume is deleted
backup = self.cinder_volume_backups.first()
api.cinder.volume_backup_get(IsA(http.HttpRequest), backup.id). \
AndReturn(backup)
api.cinder.volume_get(IsA(http.HttpRequest), backup.volume_id). \
AndRaise(self.exceptions.cinder)
self.mox.ReplayAll()
self.mock_volume_backup_get.return_value = backup
self.mock_volume_get.side_effect = self.exceptions.cinder
url = reverse('horizon:project:backups:detail',
args=[backup.id])
@ -249,20 +253,19 @@ class VolumeBackupsViewTests(test.TestCase):
self.assertTemplateUsed(res, 'horizon/common/_detail.html')
self.assertEqual(res.context['backup'].id, backup.id)
self.mock_volume_backup_get.assert_called_once_with(
test.IsHttpRequest(), backup.id)
self.mock_volume_get.assert_called_once_with(
test.IsHttpRequest(), backup.volume_id)
@test.create_stubs({api.cinder: ('volume_list',
'volume_backup_restore',)})
@test.create_mocks({api.cinder: ('volume_list',
'volume_backup_restore')})
def test_restore_backup(self):
backup = self.cinder_volume_backups.first()
volumes = self.cinder_volumes.list()
api.cinder.volume_list(IsA(http.HttpRequest)). \
AndReturn(volumes)
api.cinder.volume_backup_restore(IsA(http.HttpRequest),
backup.id,
backup.volume_id). \
AndReturn(backup)
self.mox.ReplayAll()
self.mock_volume_list.return_value = volumes
self.mock_volume_backup_restore.return_value = backup
formData = {'method': 'RestoreBackupForm',
'backup_id': backup.id,
@ -278,3 +281,6 @@ class VolumeBackupsViewTests(test.TestCase):
self.assertMessageCount(info=1)
self.assertRedirectsNoFollow(res,
reverse('horizon:project:volumes:index'))
self.mock_volume_list.assert_called_once_with(test.IsHttpRequest())
self.mock_volume_backup_restore.assert_called_once_with(
test.IsHttpRequest(), backup.id, backup.volume_id)

298
openstack_dashboard/dashboards/project/overview/tests.py
View File

@ -19,12 +19,11 @@
import datetime
import logging
from django import http
from django.test.utils import override_settings
from django.urls import reverse
from django.utils import timezone
from mox3.mox import IsA
import mock
from openstack_dashboard import api
from openstack_dashboard.test import helpers as test
@ -36,52 +35,87 @@ INDEX_URL = reverse('horizon:project:overview:index')
class UsageViewTests(test.TestCase):
@test.create_stubs({api.nova: ('usage_get',
'tenant_absolute_limits',
'extension_supported')})
def _stub_nova_api_calls(self, nova_stu_enabled=True,
@test.create_mocks({api.nova: (
'usage_get',
('tenant_absolute_limits', 'nova_tenant_absolute_limits'),
'extension_supported',
)})
def _stub_nova_api_calls(self,
nova_stu_enabled=True,
tenant_limits_exception=False,
stu_exception=False, overview_days_range=None):
api.nova.extension_supported(
'SimpleTenantUsage', IsA(http.HttpRequest)) \
.AndReturn(nova_stu_enabled)
api.nova.extension_supported(
'SimpleTenantUsage', IsA(http.HttpRequest)) \
.AndReturn(nova_stu_enabled)
stu_exception=False, overview_days_range=1):
self.mock_extension_supported.side_effect = [nova_stu_enabled,
nova_stu_enabled]
if tenant_limits_exception:
api.nova.tenant_absolute_limits(IsA(http.HttpRequest), reserved=True)\
.AndRaise(tenant_limits_exception)
self.mock_nova_tenant_absolute_limits.side_effect = \
tenant_limits_exception
else:
api.nova.tenant_absolute_limits(IsA(http.HttpRequest), reserved=True) \
.AndReturn(self.limits['absolute'])
self.mock_nova_tenant_absolute_limits.return_value = \
self.limits['absolute']
if nova_stu_enabled:
self._nova_stu_enabled(stu_exception,
overview_days_range=overview_days_range)
@test.create_stubs({api.cinder: ('tenant_absolute_limits',)})
def _stub_cinder_api_calls(self):
api.cinder.tenant_absolute_limits(IsA(http.HttpRequest)) \
.AndReturn(self.cinder_limits['absolute'])
def _check_nova_api_calls(self,
nova_stu_enabled=True,
tenant_limits_exception=False,
stu_exception=False, overview_days_range=1):
self.assert_mock_multiple_calls_with_same_arguments(
self.mock_extension_supported, 2,
mock.call('SimpleTenantUsage', test.IsHttpRequest()))
self.mock_nova_tenant_absolute_limits.assert_called_once_with(
test.IsHttpRequest(), reserved=True)
if nova_stu_enabled:
self._check_stu_enabled(stu_exception,
overview_days_range=overview_days_range)
else:
self.mock_usage_get.assert_not_called()
@test.create_stubs({api.neutron: ('is_extension_supported',
'floating_ip_supported',
@test.create_mocks({api.cinder: (
('tenant_absolute_limits', 'cinder_tenant_absolute_limits'),
)})
def _stub_cinder_api_calls(self):
self.mock_cinder_tenant_absolute_limits.return_value = \
self.cinder_limits['absolute']
def _check_cinder_api_calls(self):
self.mock_cinder_tenant_absolute_limits.assert_called_once_with(
test.IsHttpRequest())
@test.create_mocks({api.neutron: ('security_group_list',
'tenant_floating_ip_list',
'security_group_list')})
'floating_ip_supported',
'is_extension_supported')})
def _stub_neutron_api_calls(self, neutron_sg_enabled=True):
api.neutron.is_extension_supported(
IsA(http.HttpRequest),
'security-group').AndReturn(neutron_sg_enabled)
api.neutron.floating_ip_supported(IsA(http.HttpRequest)) \
.AndReturn(True)
api.neutron.tenant_floating_ip_list(IsA(http.HttpRequest)) \
.AndReturn(self.floating_ips.list())
self.mock_is_extension_supported.return_value = neutron_sg_enabled
self.mock_floating_ip_supported.return_value = True
self.mock_tenant_floating_ip_list.return_value = \
self.floating_ips.list()
if neutron_sg_enabled:
api.neutron.security_group_list(IsA(http.HttpRequest)) \
.AndReturn(self.security_groups.list())
self.mock_security_group_list.return_value = \
self.security_groups.list()
def _check_neutron_api_calls(self, neutron_sg_enabled=True):
self.mock_is_extension_supported.assert_called_once_with(
test.IsHttpRequest(), 'security-group')
self.mock_floating_ip_supported.assert_called_once_with(
test.IsHttpRequest())
self.mock_tenant_floating_ip_list.assert_called_once_with(
test.IsHttpRequest())
if neutron_sg_enabled:
self.mock_security_group_list.assert_called_once_with(
test.IsHttpRequest())
else:
self.mock_security_group_list.assert_not_called()
def _nova_stu_enabled(self, exception=False, overview_days_range=1):
if exception:
self.mock_usage_get.side_effect = exception
else:
usage = api.nova.NovaUsage(self.usages.first())
self.mock_usage_get.return_value = usage
def _check_stu_enabled(self, exception=False, overview_days_range=1):
now = timezone.now()
if overview_days_range:
start_day = now - datetime.timedelta(days=overview_days_range)
@ -91,14 +125,8 @@ class UsageViewTests(test.TestCase):
start_day.day, 0, 0, 0, 0)
end = datetime.datetime(now.year, now.month, now.day, 23, 59, 59, 0)
if exception:
api.nova.usage_get(IsA(http.HttpRequest), self.tenant.id,
start, end) \
.AndRaise(exception)
else:
api.nova.usage_get(IsA(http.HttpRequest), self.tenant.id,
start, end) \
.AndReturn(api.nova.NovaUsage(self.usages.first()))
self.mock_usage_get.assert_called_once_with(
test.IsHttpRequest(), self.tenant.id, start, end)
def _common_assertions(self, nova_stu_enabled,
maxTotalFloatingIps=float("inf")):
@ -131,41 +159,35 @@ class UsageViewTests(test.TestCase):
overview_days_range=overview_days_range)
self._stub_neutron_api_calls()
self._stub_cinder_api_calls()
self.mox.ReplayAll()
self._common_assertions(nova_stu_enabled)
self._check_nova_api_calls(nova_stu_enabled,
overview_days_range=overview_days_range)
self._check_neutron_api_calls()
self._check_cinder_api_calls()
def test_usage_nova_network(self):
self._test_usage_nova_network(nova_stu_enabled=True)
def test_usage_nova_network_disabled(self):
self._test_usage_nova_network(nova_stu_enabled=False)
@test.create_stubs({api.base: ('is_service_enabled',),
@test.create_mocks({api.base: ('is_service_enabled',),
api.cinder: ('is_volume_service_enabled',)})
def _test_usage_nova_network(self, nova_stu_enabled):
self._stub_nova_api_calls(nova_stu_enabled)
api.base.is_service_enabled(IsA(http.HttpRequest), 'network') \
.MultipleTimes().AndReturn(False)
api.cinder.is_volume_service_enabled(IsA(http.HttpRequest)) \
.MultipleTimes().AndReturn(False)
self.mox.ReplayAll()
self.mock_is_service_enabled.return_value = False
self.mock_is_volume_service_enabled.return_value = False
self._common_assertions(nova_stu_enabled, maxTotalFloatingIps=10)
@test.create_stubs({api.nova: ('usage_get',
'extension_supported')})
def _stub_nova_api_calls_unauthorized(self, exception):
api.nova.extension_supported(
'SimpleTenantUsage', IsA(http.HttpRequest)) \
.AndReturn(True)
self._nova_stu_enabled(exception)
self._check_nova_api_calls(nova_stu_enabled)
self.mock_is_service_enabled.assert_called_once_with(
test.IsHttpRequest(), 'network')
self.mock_is_volume_service_enabled.assert_called_once_with(
test.IsHttpRequest())
def test_unauthorized(self):
self.mox.ReplayAll()
url = reverse('horizon:admin:volumes:index')
# Avoid the log message in the test
@ -186,47 +208,61 @@ class UsageViewTests(test.TestCase):
def test_usage_csv_disabled(self):
self._test_usage_csv(nova_stu_enabled=False)
def _test_usage_csv(self, nova_stu_enabled=True, overview_days_range=None):
def _test_usage_csv(self, nova_stu_enabled=True, overview_days_range=1):
self._stub_nova_api_calls(nova_stu_enabled,
overview_days_range=overview_days_range)
self._stub_neutron_api_calls()
self._stub_cinder_api_calls()
self.mox.ReplayAll()
res = self.client.get(reverse('horizon:project:overview:index') +
"?format=csv")
self.assertTemplateUsed(res, 'project/overview/usage.csv')
self.assertIsInstance(res.context['usage'], usage.ProjectUsage)
self._check_nova_api_calls(nova_stu_enabled,
overview_days_range=overview_days_range)
self._check_neutron_api_calls()
self._check_cinder_api_calls()
def test_usage_exception_usage(self):
self._stub_nova_api_calls(stu_exception=self.exceptions.nova)
self._stub_neutron_api_calls()
self._stub_cinder_api_calls()
self.mox.ReplayAll()
res = self.client.get(reverse('horizon:project:overview:index'))
self.assertTemplateUsed(res, 'project/overview/usage.html')
self.assertEqual(res.context['usage'].usage_list, [])
self._check_nova_api_calls(stu_exception=self.exceptions.nova)
self._check_neutron_api_calls()
self._check_cinder_api_calls()
def test_usage_exception_quota(self):
self._stub_nova_api_calls(tenant_limits_exception=self.exceptions.nova)
self._stub_neutron_api_calls()
self._stub_cinder_api_calls()
self.mox.ReplayAll()
res = self.client.get(reverse('horizon:project:overview:index'))
self.assertTemplateUsed(res, 'project/overview/usage.html')
self.assertEqual(res.context['usage'].quotas, {})
self._check_nova_api_calls(
tenant_limits_exception=self.exceptions.nova)
self._check_neutron_api_calls()
self._check_cinder_api_calls()
def test_usage_default_tenant(self):
self._stub_nova_api_calls()
self._stub_neutron_api_calls()
self._stub_cinder_api_calls()
self.mox.ReplayAll()
res = self.client.get(reverse('horizon:project:overview:index'))
self.assertTemplateUsed(res, 'project/overview/usage.html')
self.assertIsInstance(res.context['usage'], usage.ProjectUsage)
self._check_nova_api_calls()
self._check_neutron_api_calls()
self._check_cinder_api_calls()
@test.update_settings(OPENSTACK_NEUTRON_NETWORK={'enable_quotas': True})
def test_usage_with_neutron(self):
self._test_usage_with_neutron(neutron_sg_enabled=True)
@ -239,38 +275,60 @@ class UsageViewTests(test.TestCase):
def test_usage_with_neutron_floating_ip_disabled(self):
self._test_usage_with_neutron(neutron_fip_enabled=False)
@test.create_stubs({api.neutron: ('tenant_quota_get',
'is_extension_supported',
'floating_ip_supported',
'tenant_floating_ip_list',
'security_group_list')})
def _test_usage_with_neutron_prepare(self):
self._stub_nova_api_calls()
self._stub_cinder_api_calls()
def _test_usage_with_neutron(self, neutron_sg_enabled=True,
def _check_nova_cinder_calls_with_neutron_prepare(self):
self._check_nova_api_calls()
self._check_cinder_api_calls()
@test.create_mocks({api.neutron: ('tenant_quota_get',
'is_extension_supported',
'floating_ip_supported',
'tenant_floating_ip_list',
'security_group_list')})
def _test_usage_with_neutron(self,
neutron_sg_enabled=True,
neutron_fip_enabled=True):
self._test_usage_with_neutron_prepare()
api.neutron.is_extension_supported(
IsA(http.HttpRequest), 'quotas').AndReturn(True)
api.neutron.is_extension_supported(
IsA(http.HttpRequest),
'security-group').AndReturn(neutron_sg_enabled)
api.neutron.floating_ip_supported(IsA(http.HttpRequest)) \
.AndReturn(neutron_fip_enabled)
self.mock_is_extension_supported.side_effect = [True,
neutron_sg_enabled]
self.mock_floating_ip_supported.return_value = neutron_fip_enabled
if neutron_fip_enabled:
api.neutron.tenant_floating_ip_list(IsA(http.HttpRequest)) \
.AndReturn(self.floating_ips.list())
self.mock_tenant_floating_ip_list.return_value = \
self.floating_ips.list()
if neutron_sg_enabled:
api.neutron.security_group_list(IsA(http.HttpRequest)) \
.AndReturn(self.security_groups.list())
api.neutron.tenant_quota_get(IsA(http.HttpRequest), self.tenant.id) \
.AndReturn(self.neutron_quotas.first())
self.mox.ReplayAll()
self.mock_security_group_list.return_value = \
self.security_groups.list()
self.mock_tenant_quota_get.return_value = self.neutron_quotas.first()
self._test_usage_with_neutron_check(neutron_sg_enabled,
neutron_fip_enabled)
self.mock_is_extension_supported.assert_has_calls([
mock.call(test.IsHttpRequest(), 'quotas'),
mock.call(test.IsHttpRequest(), 'security-group'),
])
self.assertEqual(2, self.mock_is_extension_supported.call_count)
self.mock_floating_ip_supported.assert_called_once_with(
test.IsHttpRequest())
if neutron_fip_enabled:
self.mock_tenant_floating_ip_list.assert_called_once_with(
test.IsHttpRequest())
else:
self.mock_tenant_floating_ip_list.assert_not_called()
if neutron_sg_enabled:
self.mock_security_group_list.assert_called_once_with(
test.IsHttpRequest())
else:
self.mock_security_group_list.assert_not_called()
self.mock_tenant_quota_get.assert_called_once_with(
test.IsHttpRequest(), self.tenant.id)
self._check_nova_cinder_calls_with_neutron_prepare()
def _test_usage_with_neutron_check(self, neutron_sg_enabled=True,
neutron_fip_expected=True,
max_fip_expected=50,
@ -290,47 +348,57 @@ class UsageViewTests(test.TestCase):
self.assertEqual(max_security_groups, max_sg_expected)
@test.update_settings(OPENSTACK_NEUTRON_NETWORK={'enable_quotas': True})
def test_usage_with_neutron_quotas_ext_error(self):
@mock.patch.object(api.neutron, 'is_extension_supported')
def test_usage_with_neutron_quotas_ext_error(self,
mock_is_extension_supported):
self._test_usage_with_neutron_prepare()
api.neutron.is_extension_supported(
IsA(http.HttpRequest), 'quotas').AndRaise(self.exceptions.neutron)
self.mox.ReplayAll()
mock_is_extension_supported.side_effect = self.exceptions.neutron
self._test_usage_with_neutron_check(neutron_fip_expected=False,
max_fip_expected=float("inf"),
max_sg_expected=float("inf"))
mock_is_extension_supported.assert_called_once_with(
test.IsHttpRequest(), 'quotas')
self._check_nova_cinder_calls_with_neutron_prepare()
@test.update_settings(OPENSTACK_NEUTRON_NETWORK={'enable_quotas': True})
def test_usage_with_neutron_sg_ext_error(self):
@mock.patch.object(api.neutron, 'is_extension_supported')
def test_usage_with_neutron_sg_ext_error(self,
mock_is_extension_supported):
self._test_usage_with_neutron_prepare()
api.neutron.is_extension_supported(
IsA(http.HttpRequest), 'quotas').AndReturn(True)
api.neutron.is_extension_supported(
IsA(http.HttpRequest),
'security-group').AndRaise(self.exceptions.neutron)
self.mox.ReplayAll()
mock_is_extension_supported.side_effect = [
True, # quotas
self.exceptions.neutron, # security-group
]
self._test_usage_with_neutron_check(neutron_fip_expected=False,
max_fip_expected=float("inf"),
max_sg_expected=float("inf"))
self.assertEqual(2, mock_is_extension_supported.call_count)
mock_is_extension_supported.assert_has_calls([
mock.call(test.IsHttpRequest(), 'quotas'),
mock.call(test.IsHttpRequest(), 'security-group'),
])
self._check_nova_cinder_calls_with_neutron_prepare()
def test_usage_with_cinder(self):
self._test_usage_cinder(cinder_enabled=True)
def test_usage_without_cinder(self):
self._test_usage_cinder(cinder_enabled=False)
@test.create_stubs({api.base: ('is_service_enabled',),
@test.create_mocks({api.base: ('is_service_enabled',),
api.cinder: ('is_volume_service_enabled',)})
def _test_usage_cinder(self, cinder_enabled):
self._stub_nova_api_calls(True)
self._stub_nova_api_calls(nova_stu_enabled=True)
if cinder_enabled:
self._stub_cinder_api_calls()
api.base.is_service_enabled(IsA(http.HttpRequest), 'network') \
.MultipleTimes().AndReturn(False)
api.cinder.is_volume_service_enabled(IsA(http.HttpRequest)) \
.MultipleTimes().AndReturn(cinder_enabled)
self.mox.ReplayAll()
self.mock_is_service_enabled.return_value = False
self.mock_is_volume_service_enabled.return_value = cinder_enabled
res = self.client.get(reverse('horizon:project:overview:index'))
usages = res.context['usage']
@ -344,13 +412,27 @@ class UsageViewTests(test.TestCase):
else:
self.assertNotIn('totalVolumesUsed', usages.limits)
self._check_nova_api_calls(nova_stu_enabled=True)
if cinder_enabled:
self._check_cinder_api_calls()
self.mock_is_service_enabled.assert_called_once_with(
test.IsHttpRequest(), 'network')
self.mock_is_volume_service_enabled.assert_called_once_with(
test.IsHttpRequest())
def _test_usage_charts(self):
self._stub_nova_api_calls(False)
self._stub_nova_api_calls(nova_stu_enabled=False)
self._stub_neutron_api_calls()
self._stub_cinder_api_calls()
self.mox.ReplayAll()
return self.client.get(reverse('horizon:project:overview:index'))
res = self.client.get(reverse('horizon:project:overview:index'))
self._check_nova_api_calls(nova_stu_enabled=False)
self._check_neutron_api_calls()
self._check_cinder_api_calls()
return res
def test_usage_charts_created(self):
res = self._test_usage_charts()

184
openstack_dashboard/dashboards/project/snapshots/tests.py
View File

@ -17,14 +17,12 @@
# under the License.
from django.conf import settings
from django import http
from django.test.utils import override_settings
from django.urls import reverse
from django.utils.http import urlunquote
from mox3.mox import IsA
import mock
from openstack_dashboard import api
from openstack_dashboard.api import cinder
from openstack_dashboard.dashboards.project.snapshots \
import tables as snapshot_tables
from openstack_dashboard.test import helpers as test
@ -35,28 +33,29 @@ INDEX_URL = reverse('horizon:project:snapshots:index')
class VolumeSnapshotsViewTests(test.TestCase):
@test.create_stubs({api.cinder: ('tenant_absolute_limits',
'volume_snapshot_list_paged',
'volume_list',),
@test.create_mocks({api.cinder: ('volume_snapshot_list_paged',
'volume_list'),
api.base: ('is_service_enabled',)})
def _test_snapshots_index_paginated(self, marker, sort_dir, snapshots, url,
has_more, has_prev):
api.base.is_service_enabled(IsA(http.HttpRequest), 'volumev3') \
.AndReturn(True)
api.base.is_service_enabled(IsA(http.HttpRequest), 'volumev3') \
.AndReturn(True)
api.cinder.volume_snapshot_list_paged(
IsA(http.HttpRequest), marker=marker, sort_dir=sort_dir,
paginate=True).AndReturn([snapshots, has_more, has_prev])
api.cinder.volume_list(IsA(http.HttpRequest)).AndReturn(
self.cinder_volumes.list())
self.mox.ReplayAll()
self.mock_is_service_enabled.return_value = True
self.mock_volume_snapshot_list_paged.return_value = [snapshots,
has_more,
has_prev]
self.mock_volume_list.return_value = self.cinder_volumes.list()
res = self.client.get(urlunquote(url))
self.assertEqual(res.status_code, 200)
self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')
self.mox.UnsetStubs()
self.assert_mock_multiple_calls_with_same_arguments(
self.mock_is_service_enabled, 2,
mock.call(test.IsHttpRequest(), 'volumev3'))
self.mock_volume_snapshot_list_paged.assert_called_once_with(
test.IsHttpRequest(), marker=marker, sort_dir=sort_dir,
paginate=True)
self.mock_volume_list.assert_called_once_with(test.IsHttpRequest())
return res
@override_settings(API_RESULT_PAGE_SIZE=1)
@ -122,42 +121,36 @@ class VolumeSnapshotsViewTests(test.TestCase):
snapshots = res.context['volume_snapshots_table'].data
self.assertItemsEqual(snapshots, expected_snapshots)
@test.create_stubs({cinder: ('volume_get',),
@test.create_mocks({api.cinder: ('volume_get',),
quotas: ('tenant_limit_usages',)})
def test_create_snapshot_get(self):
volume = self.cinder_volumes.first()
cinder.volume_get(IsA(http.HttpRequest), volume.id) \
.AndReturn(volume)
self.mock_volume_get.return_value = volume
snapshot_used = len(self.cinder_volume_snapshots.list())
usage_limit = {'maxTotalVolumeGigabytes': 250,
'totalGigabytesUsed': 20,
'totalSnapshotsUsed': snapshot_used,
'maxTotalSnapshots': 6}
quotas.tenant_limit_usages(IsA(http.HttpRequest)).\
AndReturn(usage_limit)
self.mox.ReplayAll()
self.mock_tenant_limit_usages.return_value = usage_limit
url = reverse('horizon:project:volumes:create_snapshot',
args=[volume.id])
res = self.client.get(url)
self.assertTemplateUsed(res, 'project/volumes/create_snapshot.html')
self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),
volume.id)
self.mock_tenant_limit_usages.assert_called_once_with(
test.IsHttpRequest())
@test.create_stubs({cinder: ('volume_get',
'volume_snapshot_create',)})
@test.create_mocks({api.cinder: ('volume_get',
'volume_snapshot_create')})
def test_create_snapshot_post(self):
volume = self.cinder_volumes.first()
snapshot = self.cinder_volume_snapshots.first()
cinder.volume_get(IsA(http.HttpRequest), volume.id) \
.AndReturn(volume)
cinder.volume_snapshot_create(IsA(http.HttpRequest),
volume.id,
snapshot.name,
snapshot.description,
force=False) \
.AndReturn(snapshot)
self.mox.ReplayAll()
self.mock_volume_get.return_value = volume
self.mock_volume_snapshot_create.return_value = snapshot
formData = {'method': 'CreateSnapshotForm',
'tenant_id': self.tenant.id,
@ -169,21 +162,23 @@ class VolumeSnapshotsViewTests(test.TestCase):
res = self.client.post(url, formData)
self.assertRedirectsNoFollow(res, INDEX_URL)
@test.create_stubs({cinder: ('volume_get',
'volume_snapshot_create',)})
self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),
volume.id)
self.mock_volume_snapshot_create.assert_called_once_with(
test.IsHttpRequest(),
volume.id,
snapshot.name,
snapshot.description,
force=False)
@test.create_mocks({api.cinder: ('volume_get',
'volume_snapshot_create')})
def test_force_create_snapshot(self):
volume = self.cinder_volumes.get(name='my_volume')
snapshot = self.cinder_volume_snapshots.first()
cinder.volume_get(IsA(http.HttpRequest), volume.id) \
.AndReturn(volume)
cinder.volume_snapshot_create(IsA(http.HttpRequest),
volume.id,
snapshot.name,
snapshot.description,
force=True) \
.AndReturn(snapshot)
self.mox.ReplayAll()
self.mock_volume_get.return_value = volume
self.mock_volume_snapshot_create.return_value = snapshot
formData = {'method': 'CreateSnapshotForm',
'tenant_id': self.tenant.id,
@ -195,23 +190,27 @@ class VolumeSnapshotsViewTests(test.TestCase):
res = self.client.post(url, formData)
self.assertRedirectsNoFollow(res, INDEX_URL)
@test.create_stubs({api.cinder: ('volume_snapshot_list_paged',
self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),
volume.id)
self.mock_volume_snapshot_create.assert_called_once_with(
test.IsHttpRequest(),
volume.id,
snapshot.name,
snapshot.description,
force=True)
@test.create_mocks({api.cinder: ('volume_snapshot_list_paged',
'volume_list',
'volume_snapshot_delete',
'tenant_absolute_limits')})
'volume_snapshot_delete')})
def test_delete_volume_snapshot(self):
vol_snapshots = self.cinder_volume_snapshots.list()
volumes = self.cinder_volumes.list()
snapshot = self.cinder_volume_snapshots.first()
api.cinder.volume_snapshot_list_paged(
IsA(http.HttpRequest), paginate=True, marker=None,
sort_dir='desc').AndReturn([vol_snapshots, False, False])
api.cinder.volume_list(IsA(http.HttpRequest)). \
AndReturn(volumes)
api.cinder.volume_snapshot_delete(IsA(http.HttpRequest), snapshot.id)
self.mox.ReplayAll()
self.mock_volume_snapshot_list_paged.return_value = [vol_snapshots,
False, False]
self.mock_volume_list.return_value = volumes
self.mock_volume_snapshot_delete.return_value = None
formData = {'action': 'volume_snapshots__delete__%s' % snapshot.id}
res = self.client.post(INDEX_URL, formData)
@ -219,19 +218,20 @@ class VolumeSnapshotsViewTests(test.TestCase):
self.assertRedirectsNoFollow(res, INDEX_URL)
self.assertMessageCount(success=1)
@test.create_stubs({api.cinder: ('volume_snapshot_get', 'volume_get')})
self.mock_volume_snapshot_list_paged.assert_called_once_with(
test.IsHttpRequest(), paginate=True, marker=None, sort_dir='desc')
self.mock_volume_list.assert_called_once_with(test.IsHttpRequest())
self.mock_volume_snapshot_delete.assert_called_once_with(
test.IsHttpRequest(), snapshot.id)
@test.create_mocks({api.cinder: ('volume_get',
'volume_snapshot_get')})
def test_volume_snapshot_detail_get(self):
volume = self.cinder_volumes.first()
snapshot = self.cinder_volume_snapshots.first()
api.cinder.volume_get(IsA(http.HttpRequest), volume.id). \
AndReturn(volume)
api.cinder.volume_snapshot_get(IsA(http.HttpRequest), snapshot.id). \
AndReturn(snapshot)
api.cinder.volume_get(IsA(http.HttpRequest), snapshot.volume_id). \
AndReturn(volume)
self.mox.ReplayAll()
self.mock_volume_get.return_value = volume
self.mock_volume_snapshot_get.return_value = snapshot
url = reverse('horizon:project:snapshots:detail',
args=[snapshot.id])
@ -240,14 +240,20 @@ class VolumeSnapshotsViewTests(test.TestCase):
self.assertTemplateUsed(res, 'horizon/common/_detail.html')
self.assertEqual(res.context['snapshot'].id, snapshot.id)
@test.create_stubs({api.cinder: ('volume_snapshot_get',)})
self.mock_volume_get.assert_has_calls([
mock.call(test.IsHttpRequest(), volume.id),
mock.call(test.IsHttpRequest(), snapshot.volume_id),
])
self.assertEqual(2, self.mock_volume_get.call_count)
self.mock_volume_snapshot_get.assert_called_once_with(
test.IsHttpRequest(), snapshot.id)
@test.create_mocks({api.cinder: ('volume_snapshot_get',)})
def test_volume_snapshot_detail_get_with_exception(self):
# Test to verify redirect if get volume snapshot fails
snapshot = self.cinder_volume_snapshots.first()
api.cinder.volume_snapshot_get(IsA(http.HttpRequest), snapshot.id).\
AndRaise(self.exceptions.cinder)
self.mox.ReplayAll()
self.mock_volume_snapshot_get.side_effect = self.exceptions.cinder
url = reverse('horizon:project:snapshots:detail',
args=[snapshot.id])
@ -255,38 +261,36 @@ class VolumeSnapshotsViewTests(test.TestCase):
self.assertRedirectsNoFollow(res, INDEX_URL)
@test.create_stubs({api.cinder: ('volume_snapshot_get', 'volume_get')})
self.mock_volume_snapshot_get.assert_called_once_with(
test.IsHttpRequest(), snapshot.id)
@test.create_mocks({api.cinder: ('volume_get',
'volume_snapshot_get')})
def test_volume_snapshot_detail_with_volume_get_exception(self):
# Test to verify redirect if get volume fails
volume = self.cinder_volumes.first()
snapshot = self.cinder_volume_snapshots.first()
api.cinder.volume_get(IsA(http.HttpRequest), volume.id). \
AndRaise(self.exceptions.cinder)
api.cinder.volume_snapshot_get(IsA(http.HttpRequest), snapshot.id). \
AndReturn(snapshot)
self.mox.ReplayAll()
self.mock_volume_get.side_effect = self.exceptions.cinder
self.mock_volume_snapshot_get.return_value = snapshot
url = reverse('horizon:project:snapshots:detail',
args=[snapshot.id])
res = self.client.get(url)
self.assertRedirectsNoFollow(res, INDEX_URL)
self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),
volume.id)
self.mock_volume_snapshot_get.assert_called_once_with(
test.IsHttpRequest(), snapshot.id)
@test.create_stubs({cinder: ('volume_snapshot_update',
'volume_snapshot_get')})
@test.create_mocks({api.cinder: ('volume_snapshot_update',
'volume_snapshot_get')})
def test_update_snapshot(self):
snapshot = self.cinder_volume_snapshots.first()
cinder.volume_snapshot_get(IsA(http.HttpRequest), snapshot.id) \
.AndReturn(snapshot)
cinder.volume_snapshot_update(IsA(http.HttpRequest),
snapshot.id,
snapshot.name,
snapshot.description) \
.AndReturn(snapshot)
self.mox.ReplayAll()
self.mock_volume_snapshot_get.return_value = snapshot
self.mock_volume_snapshot_update.return_value = snapshot
formData = {'method': 'UpdateSnapshotForm',
'name': snapshot.name,
@ -295,3 +299,11 @@ class VolumeSnapshotsViewTests(test.TestCase):
args=[snapshot.id])
res = self.client.post(url, formData)
self.assertRedirectsNoFollow(res, INDEX_URL)
self.mock_volume_snapshot_get.assert_called_once_with(
test.IsHttpRequest(), snapshot.id)
self.mock_volume_snapshot_update.assert_called_once_with(
test.IsHttpRequest(),
snapshot.id,
snapshot.name,
snapshot.description)