openstack
/
cinder
Code Issues Proposed changes

[1/11] Refactor test_volume file 

Due to the situation, that we've already had more that 7000 lines of
code in this file and it contains different test cases, it is
reasonably to divide it into smaller files.

This patch includes test cases that are connected with:
1) Availability zone;
2) Policy

Change-Id: I69ef0601bceb857565f36907eb8811379a5bfb77
This commit is contained in:
Mykhailo Dovgal 2017-02-08 15:40:27 +02:00 committed by Michael Dovgal
parent 693ced05f5
commit 7b5091c3ef
3 changed files with 160 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
cinder/tests/unit/test_volume.py
View File

@ -125,87 +125,6 @@ class FakeImageService(object):
'status': 'active'}
class AvailabilityZoneTestCase(base.BaseVolumeTestCase):
def setUp(self):
super(AvailabilityZoneTestCase, self).setUp()
self.get_all = self.patch(
'cinder.db.service_get_all', autospec=True,
return_value = [{'availability_zone': 'a', 'disabled': False}])
def test_list_availability_zones_cached(self):
azs = self.volume_api.list_availability_zones(enable_cache=True)
self.assertEqual([{"name": 'a', 'available': True}], list(azs))
self.assertIsNotNone(self.volume_api.availability_zones_last_fetched)
self.assertTrue(self.get_all.called)
self.volume_api.list_availability_zones(enable_cache=True)
self.assertEqual(1, self.get_all.call_count)
def test_list_availability_zones_no_cached(self):
azs = self.volume_api.list_availability_zones(enable_cache=False)
self.assertEqual([{"name": 'a', 'available': True}], list(azs))
self.assertIsNone(self.volume_api.availability_zones_last_fetched)
self.get_all.return_value[0]['disabled'] = True
azs = self.volume_api.list_availability_zones(enable_cache=False)
self.assertEqual([{"name": 'a', 'available': False}], list(azs))
self.assertIsNone(self.volume_api.availability_zones_last_fetched)
@mock.patch('oslo_utils.timeutils.utcnow')
def test_list_availability_zones_refetched(self, mock_utcnow):
mock_utcnow.return_value = datetime.datetime.utcnow()
azs = self.volume_api.list_availability_zones(enable_cache=True)
self.assertEqual([{"name": 'a', 'available': True}], list(azs))
self.assertIsNotNone(self.volume_api.availability_zones_last_fetched)
last_fetched = self.volume_api.availability_zones_last_fetched
self.assertTrue(self.get_all.called)
self.volume_api.list_availability_zones(enable_cache=True)
self.assertEqual(1, self.get_all.call_count)
# The default cache time is 3600, push past that...
mock_utcnow.return_value = (timeutils.utcnow() +
datetime.timedelta(0, 3800))
self.get_all.return_value = [
{
'availability_zone': 'a',
'disabled': False,
},
{
'availability_zone': 'b',
'disabled': False,
},
]
azs = self.volume_api.list_availability_zones(enable_cache=True)
azs = sorted([n['name'] for n in azs])
self.assertEqual(['a', 'b'], azs)
self.assertEqual(2, self.get_all.call_count)
self.assertGreater(self.volume_api.availability_zones_last_fetched,
last_fetched)
mock_utcnow.assert_called_with()
def test_list_availability_zones_enabled_service(self):
def sort_func(obj):
return obj['name']
self.get_all.return_value = [
{'availability_zone': 'ping', 'disabled': 0},
{'availability_zone': 'ping', 'disabled': 1},
{'availability_zone': 'pong', 'disabled': 0},
{'availability_zone': 'pung', 'disabled': 1},
]
volume_api = cinder.volume.api.API()
azs = volume_api.list_availability_zones()
azs = sorted(azs, key=sort_func)
expected = sorted([
{'name': 'pung', 'available': False},
{'name': 'pong', 'available': True},
{'name': 'ping', 'available': True},
], key=sort_func)
self.assertEqual(expected, azs)
@ddt.ddt
class VolumeTestCase(base.BaseVolumeTestCase):
@ -7165,39 +7084,6 @@ class FibreChannelTestCase(DriverTestCase):
self.volume.driver.validate_connector, connector)
class VolumePolicyTestCase(test.TestCase):
def setUp(self):
super(VolumePolicyTestCase, self).setUp()
cinder.policy.init()
self.context = context.get_admin_context()
def test_check_policy(self):
target = {
'project_id': self.context.project_id,
'user_id': self.context.user_id,
}
with mock.patch.object(cinder.policy, 'enforce') as mock_enforce:
cinder.volume.api.check_policy(self.context, 'attach')
mock_enforce.assert_called_once_with(self.context,
'volume:attach',
target)
def test_check_policy_with_target(self):
target = {
'project_id': self.context.project_id,
'user_id': self.context.user_id,
'id': 2,
}
with mock.patch.object(cinder.policy, 'enforce') as mock_enforce:
cinder.volume.api.check_policy(self.context, 'attach', {'id': 2})
mock_enforce.assert_called_once_with(self.context,
'volume:attach',
target)
class ImageVolumeCacheTestCase(base.BaseVolumeTestCase):
def setUp(self):

105
cinder/tests/unit/volume/test_availability_zone.py Normal file
View File

@ -0,0 +1,105 @@
# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Test for volume availability zone."""
import datetime
import mock
from oslo_utils import timeutils
from cinder.tests.unit import volume as base
import cinder.volume
class AvailabilityZoneTestCase(base.BaseVolumeTestCase):
def setUp(self):
super(AvailabilityZoneTestCase, self).setUp()
self.get_all = self.patch(
'cinder.db.service_get_all', autospec=True,
return_value = [{'availability_zone': 'a', 'disabled': False}])
def test_list_availability_zones_cached(self):
azs = self.volume_api.list_availability_zones(enable_cache=True)
self.assertEqual([{"name": 'a', 'available': True}], list(azs))
self.assertIsNotNone(self.volume_api.availability_zones_last_fetched)
self.assertTrue(self.get_all.called)
self.volume_api.list_availability_zones(enable_cache=True)
self.assertEqual(1, self.get_all.call_count)
def test_list_availability_zones_no_cached(self):
azs = self.volume_api.list_availability_zones(enable_cache=False)
self.assertEqual([{"name": 'a', 'available': True}], list(azs))
self.assertIsNone(self.volume_api.availability_zones_last_fetched)
self.get_all.return_value[0]['disabled'] = True
azs = self.volume_api.list_availability_zones(enable_cache=False)
self.assertEqual([{"name": 'a', 'available': False}], list(azs))
self.assertIsNone(self.volume_api.availability_zones_last_fetched)
@mock.patch('oslo_utils.timeutils.utcnow')
def test_list_availability_zones_refetched(self, mock_utcnow):
mock_utcnow.return_value = datetime.datetime.utcnow()
azs = self.volume_api.list_availability_zones(enable_cache=True)
self.assertEqual([{"name": 'a', 'available': True}], list(azs))
self.assertIsNotNone(self.volume_api.availability_zones_last_fetched)
last_fetched = self.volume_api.availability_zones_last_fetched
self.assertTrue(self.get_all.called)
self.volume_api.list_availability_zones(enable_cache=True)
self.assertEqual(1, self.get_all.call_count)
# The default cache time is 3600, push past that...
mock_utcnow.return_value = (timeutils.utcnow() +
datetime.timedelta(0, 3800))
self.get_all.return_value = [
{
'availability_zone': 'a',
'disabled': False,
},
{
'availability_zone': 'b',
'disabled': False,
},
]
azs = self.volume_api.list_availability_zones(enable_cache=True)
azs = sorted([n['name'] for n in azs])
self.assertEqual(['a', 'b'], azs)
self.assertEqual(2, self.get_all.call_count)
self.assertGreater(self.volume_api.availability_zones_last_fetched,
last_fetched)
mock_utcnow.assert_called_with()
def test_list_availability_zones_enabled_service(self):
def sort_func(obj):
return obj['name']
self.get_all.return_value = [
{'availability_zone': 'ping', 'disabled': 0},
{'availability_zone': 'ping', 'disabled': 1},
{'availability_zone': 'pong', 'disabled': 0},
{'availability_zone': 'pung', 'disabled': 1},
]
volume_api = cinder.volume.api.API()
azs = volume_api.list_availability_zones()
azs = sorted(azs, key=sort_func)
expected = sorted([
{'name': 'pung', 'available': False},
{'name': 'pong', 'available': True},
{'name': 'ping', 'available': True},
], key=sort_func)
self.assertEqual(expected, azs)

55
cinder/tests/unit/volume/test_policy.py Normal file
View File

@ -0,0 +1,55 @@
# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# All Rights Reserved.
#
# 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.
"""Tests for volume policy."""
import mock
from cinder import context
from cinder import test
import cinder.policy
class VolumePolicyTestCase(test.TestCase):
def setUp(self):
super(VolumePolicyTestCase, self).setUp()
cinder.policy.init()
self.context = context.get_admin_context()
def test_check_policy(self):
target = {
'project_id': self.context.project_id,
'user_id': self.context.user_id,
}
with mock.patch.object(cinder.policy, 'enforce') as mock_enforce:
cinder.volume.api.check_policy(self.context, 'attach')
mock_enforce.assert_called_once_with(self.context,
'volume:attach',
target)
def test_check_policy_with_target(self):
target = {
'project_id': self.context.project_id,
'user_id': self.context.user_id,
'id': 2,
}
with mock.patch.object(cinder.policy, 'enforce') as mock_enforce:
cinder.volume.api.check_policy(self.context, 'attach', {'id': 2})
mock_enforce.assert_called_once_with(self.context,
'volume:attach',
target)